// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

// Unit tests for DataType (and subclasses), Field, and Schema

#include <algorithm>
#include <cctype>
#include <cstdint>
#include <functional>
#include <memory>
#include <string>
#include <unordered_set>
#include <vector>

#include <kmock/kmock.h>

#include <nebula/core/array.h>
#include <nebula/core/memory_pool.h>
#include <nebula/core/table.h>
#include <nebula/testing/ktest_util.h>
#include <nebula/testing/random.h>
#include <nebula/testing/util.h>
#include <nebula/types/type.h>
#include <nebula/types/type_traits.h>
#include <turbo/base/checked_cast.h>
#include <nebula/util/key_value_metadata.h>

namespace nebula {
    TEST(TestTypeId, AllTypeIds) {
        const auto all_ids = AllTypeIds();
        ASSERT_EQ(static_cast<int>(all_ids.size()), Type::MAX_ID);
    }

    template<typename ReprFunc>
    void CheckTypeIdReprs(ReprFunc &&repr_func, bool expect_uppercase) {
        std::unordered_set<std::string> unique_reprs;
        const auto all_ids = AllTypeIds();
        for (const auto id: all_ids) {
            std::string repr = repr_func(id);
            ASSERT_TRUE(std::all_of(repr.begin(), repr.end(),
                [=](const char c) {
                return c == '_' || std::isdigit(c) ||
                (expect_uppercase ? std::isupper(c)
                    : std::islower(c));
                }))
        << "Invalid type id repr: '" << repr << "'";
            unique_reprs.insert(std::move(repr));
        }
        // No duplicates
        ASSERT_EQ(unique_reprs.size(), all_ids.size());
    }

    TEST(TestTypeId, ToString) {
        // Should be all uppercase strings (corresponding to the enum member names)
        CheckTypeIdReprs([](Type::type id) { return internal::to_string(id); },
                         /* expect_uppercase=*/true);
    }

    TEST(TestTypeId, to_type_name) {
        // Should be all lowercase strings (corresponding to TypeClass::type_name())
        CheckTypeIdReprs([](Type::type id) { return internal::to_type_name(id); },
                         /* expect_uppercase=*/false);
    }

    TEST(TestField, Basics) {
        Field f0("f0", int32());
        Field f0_nn("f0", int32(), false);

        ASSERT_EQ(f0.name(), "f0");
        ASSERT_EQ(f0.type()->to_string(), int32()->to_string());

        ASSERT_TRUE(f0.nullable());
        ASSERT_FALSE(f0_nn.nullable());
    }

    TEST(TestField, ToString) {
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto f0 = field("f0", int32(), false, metadata);

        std::string result = f0->to_string(/*print_metadata=*/true);
        std::string expected = R"(f0: int32 not null
-- metadata --
foo: bizz
bar: buzz)";
        ASSERT_EQ(expected, result);

        result = f0->to_string();
        expected = "f0: int32 not null";
        ASSERT_EQ(expected, result);
    }

    TEST(TestField, Equals) {
        auto meta1 = key_value_metadata({{"a", "1"}, {"b", "2"}});
        // Different from meta1
        auto meta2 = key_value_metadata({{"a", "1"}, {"b", "3"}});
        // Equal to meta1, though in different order
        auto meta3 = key_value_metadata({{"b", "2"}, {"a", "1"}});

        Field f0("f0", int32());
        Field f0_nn("f0", int32(), false);
        Field f0_other("f0", int32());
        Field f0_with_meta1("f0", int32(), true, meta1);
        Field f0_with_meta2("f0", int32(), true, meta2);
        Field f0_with_meta3("f0", int32(), true, meta3);

        AssertFieldEqual(f0, f0_other);
        AssertFieldNotEqual(f0, f0_nn);
        AssertFieldNotEqual(f0, f0_with_meta1, /*check_metadata=*/true);
        AssertFieldNotEqual(f0_with_meta1, f0_with_meta2, /*check_metadata=*/true);
        AssertFieldEqual(f0_with_meta1, f0_with_meta3, /*check_metadata=*/true);

        AssertFieldEqual(f0, f0_with_meta1);
        AssertFieldEqual(f0, f0_with_meta2);
        AssertFieldEqual(f0_with_meta1, f0_with_meta2);

        // operator==(), where check_metadata == false
        ASSERT_EQ(f0, f0_other);
        ASSERT_NE(f0, f0_nn);
        ASSERT_EQ(f0, f0_with_meta1);
        ASSERT_EQ(f0_with_meta1, f0_with_meta2);
    }

#define ASSERT_COMPATIBLE_IMPL(NAME, TYPE, PLURAL)                        \
  void Assert##NAME##Compatible(const TYPE& left, const TYPE& right) {    \
    ASSERT_TRUE(left.is_compatible_with(right))                             \
        << PLURAL << left.to_string() << "' and '" << right.to_string()     \
        << "' should be compatible";                                      \
  }                                                                       \
                                                                          \
  void Assert##NAME##Compatible(const std::shared_ptr<TYPE>& left,        \
                                const std::shared_ptr<TYPE>& right) {     \
    ASSERT_NE(left, nullptr);                                             \
    ASSERT_NE(right, nullptr);                                            \
    Assert##NAME##Compatible(*left, *right);                              \
  }                                                                       \
                                                                          \
  void Assert##NAME##NotCompatible(const TYPE& left, const TYPE& right) { \
    ASSERT_FALSE(left.is_compatible_with(right))                            \
        << PLURAL << left.to_string() << "' and '" << right.to_string()     \
        << "' should not be compatible";                                  \
  }                                                                       \
                                                                          \
  void Assert##NAME##NotCompatible(const std::shared_ptr<TYPE>& left,     \
                                   const std::shared_ptr<TYPE>& right) {  \
    ASSERT_NE(left, nullptr);                                             \
    ASSERT_NE(right, nullptr);                                            \
    Assert##NAME##NotCompatible(*left, *right);                           \
  }

    ASSERT_COMPATIBLE_IMPL(Field, Field, "fields")
#undef ASSERT_COMPATIBLE_IMPL

    TEST(TestField, is_compatible_with) {
        auto meta1 = key_value_metadata({{"a", "1"}, {"b", "2"}});
        // Different from meta1
        auto meta2 = key_value_metadata({{"a", "1"}, {"b", "3"}});
        // Equal to meta1, though in different order
        auto meta3 = key_value_metadata({{"b", "2"}, {"a", "1"}});

        Field f0("f0", int32());
        Field f0_nn("f0", int32(), false);
        Field f0_nt("f0", null());
        Field f0_other("f0", int32());
        Field f0_with_meta1("f0", int32(), true, meta1);
        Field f0_with_meta2("f0", int32(), true, meta2);
        Field f0_with_meta3("f0", int32(), true, meta3);
        Field other("other", int64());

        AssertFieldCompatible(f0, f0_other);
        AssertFieldCompatible(f0, f0_with_meta1);
        AssertFieldCompatible(f0, f0_nn);
        AssertFieldCompatible(f0, f0_nt);
        AssertFieldCompatible(f0_nt, f0_with_meta1);
        AssertFieldCompatible(f0_with_meta1, f0_with_meta2);
        AssertFieldCompatible(f0_with_meta1, f0_with_meta3);
        AssertFieldNotCompatible(f0, other);
    }

    TEST(TestField, TestMetadataConstruction) {
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto metadata2 = metadata->copy();
        auto f0 = field("f0", int32(), true, metadata);
        auto f1 = field("f0", int32(), true, metadata2);
        ASSERT_TRUE(metadata->equals(*f0->metadata()));
        AssertFieldEqual(f0, f1);
    }

    TEST(TestField, TestWithMetadata) {
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto f0 = field("f0", int32());
        auto f1 = field("f0", int32(), true, metadata);
        std::shared_ptr<Field> f2 = f0->with_metadata(metadata);

        AssertFieldEqual(f0, f2);
        AssertFieldNotEqual(f0, f2, /*check_metadata=*/true);

        AssertFieldEqual(f1, f2);
        AssertFieldEqual(f1, f2, /*check_metadata=*/true);

        // Ensure pointer equality for zero-copy
        ASSERT_EQ(metadata.get(), f1->metadata().get());
    }

    TEST(TestField, TestWithMergedMetadata) {
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto f0 = field("f0", int32(), true, metadata);
        auto f1 = field("f0", int32());

        auto metadata2 = key_value_metadata({"bar", "baz"}, {"bozz", "bazz"});

        auto f2 = f0->with_merged_metadata(metadata2);
        auto expected = field("f0", int32(), true, metadata->Merge(*metadata2));
        AssertFieldEqual(expected, f2);

        auto f3 = f1->with_merged_metadata(metadata2);
        expected = field("f0", int32(), true, metadata2);
        AssertFieldEqual(expected, f3);
    }

    TEST(TestField, TestRemoveMetadata) {
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto f0 = field("f0", int32());
        auto f1 = field("f0", int32(), true, metadata);
        std::shared_ptr<Field> f2 = f1->remove_metadata();
        ASSERT_EQ(f2->metadata(), nullptr);
    }

    TEST(TestField, TestEmptyMetadata) {
        // Empty metadata should be equivalent to no metadata at all
        auto metadata1 = key_value_metadata({});
        auto metadata2 = key_value_metadata({"foo"}, {"foo value"});

        auto f0 = field("f0", int32());
        auto f1 = field("f0", int32(), true, metadata1);
        auto f2 = field("f0", int32(), true, metadata2);

        AssertFieldEqual(f0, f1);
        AssertFieldEqual(f0, f2);
        AssertFieldEqual(f0, f1, /*check_metadata =*/true);
        AssertFieldNotEqual(f0, f2, /*check_metadata =*/true);
    }

    TEST(TestField, TestFlatten) {
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto f0 = field("f0", int32(), true /* nullable */, metadata);
        auto vec = f0->Flatten();
        ASSERT_EQ(vec.size(), 1);
        AssertFieldEqual(vec[0], f0);

        auto f1 = field("f1", float64(), false /* nullable */);
        auto ff = field("nest", STRUCT({f0, f1}));
        vec = ff->Flatten();
        ASSERT_EQ(vec.size(), 2);
        auto expected0 = field("nest.f0", int32(), true /* nullable */, metadata);
        // nullable parent implies nullable flattened child
        auto expected1 = field("nest.f1", float64(), true /* nullable */);
        AssertFieldEqual(vec[0], expected0);
        AssertFieldEqual(vec[1], expected1);

        ff = field("nest", STRUCT({f0, f1}), false /* nullable */);
        vec = ff->Flatten();
        ASSERT_EQ(vec.size(), 2);
        expected0 = field("nest.f0", int32(), true /* nullable */, metadata);
        expected1 = field("nest.f1", float64(), false /* nullable */);
        AssertFieldEqual(vec[0], expected0);
        AssertFieldEqual(vec[1], expected1);
    }

    TEST(TestField, TestReplacement) {
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto f0 = field("f0", int32(), true, metadata);
        auto fzero = f0->with_type(utf8());
        auto f1 = f0->with_name("f1");

        AssertFieldNotEqual(f0, fzero);
        AssertFieldNotCompatible(f0, fzero);
        AssertFieldNotEqual(fzero, f1);
        AssertFieldNotCompatible(fzero, f1);
        AssertFieldNotEqual(f1, f0);
        AssertFieldNotCompatible(f1, f0);

        ASSERT_EQ(fzero->name(), "f0");
        AssertTypeEqual(fzero->type(), utf8());
        ASSERT_TRUE(fzero->metadata()->equals(*metadata));

        ASSERT_EQ(f1->name(), "f1");
        AssertTypeEqual(f1->type(), int32());
        ASSERT_TRUE(f1->metadata()->equals(*metadata));
    }

    TEST(TestField, TestMerge) {
        auto metadata1 = key_value_metadata({"foo"}, {"v"});
        auto metadata2 = key_value_metadata({"bar"}, {"v"}); {
            // different name.
            ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, field("f0", int32())->merge_with(field("f1", int32())));
        } {
            // Same type.
            auto f1 = field("f", int32())->with_metadata(metadata1);
            auto f2 = field("f", int32())->with_metadata(metadata2);
            std::shared_ptr<Field> result;
            ASSERT_OK_AND_ASSIGN(result, f1->merge_with(f2));
            ASSERT_TRUE(result->equals(f1));
            ASSERT_OK_AND_ASSIGN(result, f2->merge_with(f1));
            ASSERT_TRUE(result->equals(f2));
        } {
            // promote_nullability == false
            auto f = field("f", int32());
            auto null_field = field("f", null());
            Field::MergeOptions options;
            options.promote_nullability = false;
            ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, f->merge_with(null_field, options));
            ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, null_field->merge_with(f, options));

            // Also rejects fields with different nullability.
            ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition,
                          f->with_nullable(true)->merge_with(f->with_nullable(false), options));
        } {
            // promote_nullability == true; merge with a null field.
            Field::MergeOptions options;
            options.promote_nullability = true;
            auto f = field("f", int32())->with_nullable(false)->with_metadata(metadata1);
            auto null_field = field("f", null())->with_metadata(metadata2);

            std::shared_ptr<Field> result;
            ASSERT_OK_AND_ASSIGN(result, f->merge_with(null_field, options));
            ASSERT_TRUE(result->equals(f->with_nullable(true)->with_metadata(metadata1)));
            ASSERT_OK_AND_ASSIGN(result, null_field->merge_with(f, options));
            ASSERT_TRUE(result->equals(f->with_nullable(true)->with_metadata(metadata2)));
        } {
            // promote_nullability == true; merge a nullable field and an in-nullable field.
            Field::MergeOptions options;
            options.promote_nullability = true;
            auto f1 = field("f", int32())->with_nullable(false);
            auto f2 = field("f", int32())->with_nullable(true);
            std::shared_ptr<Field> result;
            ASSERT_OK_AND_ASSIGN(result, f1->merge_with(f2, options));
            ASSERT_TRUE(result->equals(f1->with_nullable(true)));
            ASSERT_OK_AND_ASSIGN(result, f2->merge_with(f1, options));
            ASSERT_TRUE(result->equals(f2));
        }
    }

    using TestSchema = ::testing::Test;

    TEST_F(TestSchema, Basics) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f1_optional = field("f1", uint8());

        auto f2 = field("f2", utf8());

        auto schema = ::nebula::schema({f0, f1, f2});

        ASSERT_EQ(3, schema->num_fields());
        AssertFieldEqual(*f0, *schema->field(0));
        AssertFieldEqual(*f1, *schema->field(1));
        AssertFieldEqual(*f2, *schema->field(2));

        auto schema2 = ::nebula::schema({f0, f1, f2});

        std::vector<std::shared_ptr<Field> > fields3 = {f0, f1_optional, f2};
        auto schema3 = std::make_shared<Schema>(fields3);
        AssertSchemaEqual(schema, schema2);
        AssertSchemaNotEqual(schema, schema3);
        ASSERT_EQ(*schema, *schema2);
        ASSERT_NE(*schema, *schema3);

        ASSERT_EQ(schema->fingerprint(), schema2->fingerprint());
        ASSERT_NE(schema->fingerprint(), schema3->fingerprint());

        auto schema4 = ::nebula::schema({f0}, Endianness::Little);
        auto schema5 = ::nebula::schema({f0}, Endianness::Little);
        auto schema6 = ::nebula::schema({f0}, Endianness::Big);
        auto schema7 = ::nebula::schema({f0});

        AssertSchemaEqual(schema4, schema5);
        AssertSchemaNotEqual(schema4, schema6);
#if NEBULA_LITTLE_ENDIAN
        AssertSchemaEqual(schema4, schema7);
        AssertSchemaNotEqual(schema6, schema7);
#else
        AssertSchemaNotEqual(schema4, schema6);
        AssertSchemaEqual(schema6, schema7);
#endif

        ASSERT_EQ(schema4->fingerprint(), schema5->fingerprint());
        ASSERT_NE(schema4->fingerprint(), schema6->fingerprint());
#if NEBULA_LITTLE_ENDIAN
        ASSERT_EQ(schema4->fingerprint(), schema7->fingerprint());
        ASSERT_NE(schema6->fingerprint(), schema7->fingerprint());
#else
        ASSERT_NE(schema4->fingerprint(), schema7->fingerprint());
        ASSERT_EQ(schema6->fingerprint(), schema7->fingerprint());
#endif

        auto schema8 = ::nebula::schema({field("f0", int8()), field("f1", int32())});
        auto schema9 = ::nebula::schema({{"f0", int8()}, {"f1", int32()}});
        auto schema10 = ::nebula::schema({{"f2", int8()}, {"f1", int32()}});

        AssertSchemaEqual(schema8, schema9);
        AssertSchemaNotEqual(schema8, schema10);
    }

    TEST_F(TestSchema, ToString) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f3", list(int16()));

        auto metadata = key_value_metadata({"foo"}, {"bar"});
        auto schema = ::nebula::schema({f0, f1, f2, f3}, metadata);

        std::string result = schema->to_string();
        std::string expected = R"(f0: int32
f1: uint8 not null
f2: string
f3: list<item: int16>)";

        ASSERT_EQ(expected, result);

        result = schema->to_string(/*print_metadata=*/true);
        std::string expected_with_metadata = expected + R"(
-- metadata --
foo: bar)";

        ASSERT_EQ(expected_with_metadata, result);

        // With swapped endianness
#if NEBULA_LITTLE_ENDIAN
        schema = schema->with_endianness(Endianness::Big);
        expected = R"(f0: int32
f1: uint8 not null
f2: string
f3: list<item: int16>
-- endianness: big --)";
#else
        schema = schema->with_endianness(Endianness::Little);
        expected = R"(f0: int32
f1: uint8 not null
f2: string
f3: list<item: int16>
-- endianness: little --)";
#endif

        result = schema->to_string();
        ASSERT_EQ(expected, result);

        result = schema->to_string(/*print_metadata=*/true);
        expected_with_metadata = expected + R"(
-- metadata --
foo: bar)";

        ASSERT_EQ(expected_with_metadata, result);
    }

    TEST_F(TestSchema, get_field_by_name) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f3", list(int16()));

        auto schema = ::nebula::schema({f0, f1, f2, f3});

        std::shared_ptr<Field> result;

        result = schema->get_field_by_name("f1");
        AssertFieldEqual(f1, result);

        result = schema->get_field_by_name("f3");
        AssertFieldEqual(f3, result);

        result = schema->get_field_by_name("not-found");
        ASSERT_EQ(result, nullptr);
    }

    TEST_F(TestSchema, get_field_index) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f3", list(int16()));

        auto schema = ::nebula::schema({f0, f1, f2, f3});

        ASSERT_EQ(0, schema->get_field_index(f0->name()));
        ASSERT_EQ(1, schema->get_field_index(f1->name()));
        ASSERT_EQ(2, schema->get_field_index(f2->name()));
        ASSERT_EQ(3, schema->get_field_index(f3->name()));
        ASSERT_EQ(-1, schema->get_field_index("not-found"));
    }

    TEST_F(TestSchema, GetFieldDuplicates) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f1", list(int16()));

        auto schema = ::nebula::schema({f0, f1, f2, f3});

        ASSERT_EQ(0, schema->get_field_index(f0->name()));
        ASSERT_EQ(-1, schema->get_field_index(f1->name())); // duplicate
        ASSERT_EQ(2, schema->get_field_index(f2->name()));
        ASSERT_EQ(-1, schema->get_field_index("not-found"));
        ASSERT_EQ(std::vector<int>{0}, schema->get_all_field_indices(f0->name()));
        ASSERT_EQ(std::vector<int>({1, 3}), schema->get_all_field_indices(f1->name()));

        ASSERT_TRUE(::nebula::schema({f0, f1, f2})->has_distinct_field_names());
        ASSERT_FALSE(schema->has_distinct_field_names());

        std::vector<std::shared_ptr<Field> > results;

        results = schema->get_all_fields_by_name(f0->name());
        ASSERT_EQ(results.size(), 1);
        AssertFieldEqual(results[0], f0);

        results = schema->get_all_fields_by_name(f1->name());
        ASSERT_EQ(results.size(), 2);
        if (results[0]->type()->id() == Type::UINT8) {
            AssertFieldEqual(results[0], f1);
            AssertFieldEqual(results[1], f3);
        } else {
            AssertFieldEqual(results[0], f3);
            AssertFieldEqual(results[1], f1);
        }

        results = schema->get_all_fields_by_name("not-found");
        ASSERT_EQ(results.size(), 0);
    }

    TEST_F(TestSchema, can_reference_field_by_name) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f1", list(int16()));

        auto schema = ::nebula::schema({f0, f1, f2, f3});

        ASSERT_OK(schema->can_reference_field_by_name("f0"));
        ASSERT_OK(schema->can_reference_field_by_name("f2"));

        // Not found
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, schema->can_reference_field_by_name("nope"));

        // Duplicates
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, schema->can_reference_field_by_name("f1"));
    }

    TEST_F(TestSchema, can_reference_fields_by_names) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f1", list(int16()));

        auto schema = ::nebula::schema({f0, f1, f2, f3});

        ASSERT_OK(schema->can_reference_fields_by_names({"f0", "f2"}));
        ASSERT_OK(schema->can_reference_fields_by_names({"f2", "f0"}));

        // Not found
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, schema->can_reference_fields_by_names({"nope"}));
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, schema->can_reference_fields_by_names({"f0", "nope"}));
        // Duplicates
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, schema->can_reference_fields_by_names({"f1"}));
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, schema->can_reference_fields_by_names({"f0", "f1"}));
        // Both
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, schema->can_reference_fields_by_names({"f0", "f1", "nope"}));
    }

    TEST_F(TestSchema, TestMetadataConstruction) {
        auto metadata0 = key_value_metadata({{"foo", "bar"}, {"bizz", "buzz"}});
        auto metadata1 = key_value_metadata({{"foo", "baz"}});

        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8(), true);
        auto f3 = field("f2", utf8(), true, metadata1->copy());

        auto schema0 = ::nebula::schema({f0, f1, f2}, metadata0);
        auto schema1 = ::nebula::schema({f0, f1, f2}, metadata1);
        auto schema2 = ::nebula::schema({f0, f1, f2}, metadata0->copy());
        auto schema3 = ::nebula::schema({f0, f1, f3}, metadata0->copy());

        ASSERT_TRUE(metadata0->equals(*schema0->metadata()));
        ASSERT_TRUE(metadata1->equals(*schema1->metadata()));
        ASSERT_TRUE(metadata0->equals(*schema2->metadata()));
        AssertSchemaEqual(schema0, schema2);

        AssertSchemaEqual(schema0, schema1);
        AssertSchemaNotEqual(schema0, schema1, /*check_metadata=*/true);

        AssertSchemaEqual(schema2, schema1);
        AssertSchemaNotEqual(schema2, schema1, /*check_metadata=*/true);

        // Field has different metadata
        AssertSchemaEqual(schema2, schema3);
        AssertSchemaNotEqual(schema2, schema3, /*check_metadata=*/true);

        ASSERT_EQ(schema0->fingerprint(), schema1->fingerprint());
        ASSERT_EQ(schema0->fingerprint(), schema2->fingerprint());
        ASSERT_EQ(schema0->fingerprint(), schema3->fingerprint());
        ASSERT_NE(schema0->metadata_fingerprint(), schema1->metadata_fingerprint());
        ASSERT_EQ(schema0->metadata_fingerprint(), schema2->metadata_fingerprint());
        ASSERT_NE(schema0->metadata_fingerprint(), schema3->metadata_fingerprint());
    }

    TEST_F(TestSchema, TestNestedMetadataComparison) {
        auto item0 = field("item", int32(), true);
        auto item1 = field("item", int32(), true, key_value_metadata({{"foo", "baz"}}));

        Schema schema0({field("f", list(item0))});
        Schema schema1({field("f", list(item1))});

        ASSERT_EQ(schema0.fingerprint(), schema1.fingerprint());
        ASSERT_NE(schema0.metadata_fingerprint(), schema1.metadata_fingerprint());

        AssertSchemaEqual(schema0, schema1);
        AssertSchemaNotEqual(schema0, schema1, /* check_metadata = */ true);
    }

    TEST_F(TestSchema, TestDeeplyNestedMetadataComparison) {
        auto item0 = field("item", int32(), true);
        auto item1 = field("item", int32(), true, key_value_metadata({{"foo", "baz"}}));

        Schema schema0(
            {field("f", list(list(sparse_union({field("struct", STRUCT({item0}))}))))});
        Schema schema1(
            {field("f", list(list(sparse_union({field("struct", STRUCT({item1}))}))))});

        ASSERT_EQ(schema0.fingerprint(), schema1.fingerprint());
        ASSERT_NE(schema0.metadata_fingerprint(), schema1.metadata_fingerprint());

        AssertSchemaEqual(schema0, schema1);
        AssertSchemaNotEqual(schema0, schema1, /* check_metadata = */ true);
    }

    TEST_F(TestSchema, TestFieldsDifferOnlyInMetadata) {
        auto f0 = field("f", utf8(), true, nullptr);
        auto f1 = field("f", utf8(), true, key_value_metadata({{"foo", "baz"}}));

        Schema schema0({f0, f1});
        Schema schema1({f1, f0});

        AssertSchemaEqual(schema0, schema1);
        AssertSchemaNotEqual(schema0, schema1, /* check_metadata = */ true);

        ASSERT_EQ(schema0.fingerprint(), schema1.fingerprint());
        ASSERT_NE(schema0.metadata_fingerprint(), schema1.metadata_fingerprint());
    }

    TEST_F(TestSchema, TestEmptyMetadata) {
        // Empty metadata should be equivalent to no metadata at all
        auto f1 = field("f1", int32());
        auto metadata1 = key_value_metadata({});
        auto metadata2 = key_value_metadata({"foo"}, {"foo value"});

        auto schema1 = ::nebula::schema({f1});
        auto schema2 = ::nebula::schema({f1}, metadata1);
        auto schema3 = ::nebula::schema({f1}, metadata2);

        AssertSchemaEqual(schema1, schema2);
        AssertSchemaNotEqual(schema1, schema3, /*check_metadata=*/true);

        ASSERT_EQ(schema1->fingerprint(), schema2->fingerprint());
        ASSERT_EQ(schema1->fingerprint(), schema3->fingerprint());
        ASSERT_EQ(schema1->metadata_fingerprint(), schema2->metadata_fingerprint());
        ASSERT_NE(schema1->metadata_fingerprint(), schema3->metadata_fingerprint());
    }

    TEST_F(TestSchema, TestWithMetadata) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        std::vector<std::shared_ptr<Field> > fields = {f0, f1, f2};
        auto metadata = key_value_metadata({"foo", "bar"}, {"bizz", "buzz"});
        auto schema = std::make_shared<Schema>(fields);
        std::shared_ptr<Schema> new_schema = schema->with_metadata(metadata);
        ASSERT_TRUE(metadata->equals(*new_schema->metadata()));

        // Not copied
        ASSERT_TRUE(metadata.get() == new_schema->metadata().get());
    }

    TEST_F(TestSchema, TestRemoveMetadata) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        std::vector<std::shared_ptr<Field> > fields = {f0, f1, f2};
        auto schema = std::make_shared<Schema>(fields);
        std::shared_ptr<Schema> new_schema = schema->remove_metadata();
        ASSERT_TRUE(new_schema->metadata() == nullptr);
    }

    void AssertSchemaBuilderYield(const SchemaBuilder &builder,
                                  const std::shared_ptr<Schema> &expected) {
        ASSERT_OK_AND_ASSIGN(auto schema, builder.finish());
        AssertSchemaEqual(schema, expected);
    }

    TEST(TestSchemaBuilder, DefaultBehavior) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());

        SchemaBuilder builder;
        ASSERT_OK(builder.add_field(f0));
        ASSERT_OK(builder.add_field(f1));
        ASSERT_OK(builder.add_field(f2));
        AssertSchemaBuilderYield(builder, schema({f0, f1, f2}));

        builder.reset();
        ASSERT_OK(builder.add_fields({f0, f1, f2->with_nullable(false)}));
        AssertSchemaBuilderYield(builder, schema({f0, f1, f2->with_nullable(false)}));

        builder.reset();
        ASSERT_OK(builder.add_schema(schema({f2, f0})));
        AssertSchemaBuilderYield(builder, schema({f2, f0}));

        builder.reset();
        ASSERT_OK(builder.add_schemas({schema({f1, f2}), schema({f2, f0})}));
        AssertSchemaBuilderYield(builder, schema({f1, f2, f2, f0}));
    }

    TEST(TestSchemaBuilder, with_metadata) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto metadata = key_value_metadata({{"foo", "bar"}});

        SchemaBuilder builder;
        ASSERT_OK(builder.add_metadata(*metadata));
        ASSERT_OK_AND_ASSIGN(auto schema, builder.finish());
        AssertSchemaEqual(schema, ::nebula::schema({})->with_metadata(metadata));

        ASSERT_OK(builder.add_field(f0));
        ASSERT_OK_AND_ASSIGN(schema, builder.finish());
        AssertSchemaEqual(schema, ::nebula::schema({f0})->with_metadata(metadata));

        SchemaBuilder other_builder{::nebula::schema({})->with_metadata(metadata)};
        ASSERT_OK(other_builder.add_field(f1));
        ASSERT_OK_AND_ASSIGN(schema, other_builder.finish());
        AssertSchemaEqual(schema, ::nebula::schema({f1})->with_metadata(metadata));

        other_builder.reset();
        ASSERT_OK(other_builder.add_field(f1->with_metadata(metadata)));
        ASSERT_OK_AND_ASSIGN(schema, other_builder.finish());
        AssertSchemaEqual(schema, ::nebula::schema({f1->with_metadata(metadata)}));
    }

    TEST(TestSchemaBuilder, IncrementalConstruction) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());

        SchemaBuilder builder;
        std::shared_ptr<Schema> actual;

        ASSERT_OK_AND_ASSIGN(actual, builder.finish());
        AssertSchemaEqual(actual, ::nebula::schema({}));

        ASSERT_OK(builder.add_field(f0));
        ASSERT_OK_AND_ASSIGN(actual, builder.finish());
        AssertSchemaEqual(actual, ::nebula::schema({f0}));

        ASSERT_OK(builder.add_field(f1));
        ASSERT_OK_AND_ASSIGN(actual, builder.finish());
        AssertSchemaEqual(actual, ::nebula::schema({f0, f1}));

        ASSERT_OK(builder.add_field(f2));
        AssertSchemaBuilderYield(builder, schema({f0, f1, f2}));
    }

    TEST(TestSchemaBuilder, PolicyIgnore) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8());
        auto f0_req = field("f0", utf8(), false);

        SchemaBuilder builder{SchemaBuilder::CONFLICT_IGNORE};

        ASSERT_OK(builder.add_fields({f0, f1}));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));

        ASSERT_OK(builder.add_field(f0_req));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));

        ASSERT_OK(builder.add_field(f0));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));
    }

    TEST(TestSchemaBuilder, PolicyReplace) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8());
        auto f0_req = field("f0", utf8(), false);

        SchemaBuilder builder{SchemaBuilder::CONFLICT_REPLACE};

        ASSERT_OK(builder.add_fields({f0, f1}));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));

        ASSERT_OK(builder.add_field(f0_req));
        AssertSchemaBuilderYield(builder, schema({f0_req, f1}));

        ASSERT_OK(builder.add_field(f0));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));
    }

    TEST(TestSchemaBuilder, PolicyMerge) {
        auto f0 = field("f0", int32(), true);
        auto f1 = field("f1", uint8());
        // Same as f0, but not required.
        auto f0_opt = field("f0", int32());
        // Another type, can't merge
        auto f0_other = field("f0", utf8(), false);

        SchemaBuilder builder{SchemaBuilder::CONFLICT_MERGE};

        ASSERT_OK(builder.add_fields({f0, f1}));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));

        ASSERT_OK(builder.add_field(f0_opt));
        AssertSchemaBuilderYield(builder, schema({f0_opt, f1}));

        // Unsupported merge with a different type
        ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, builder.add_field(f0_other));
        // Builder should still contain state
        AssertSchemaBuilderYield(builder, schema({f0, f1}));

        builder.reset();
        // create a schema with duplicate fields
        builder.set_policy(SchemaBuilder::CONFLICT_APPEND);
        ASSERT_OK(builder.add_fields({f0, f0}));

        builder.set_policy(SchemaBuilder::CONFLICT_MERGE);
        // Even if the field is compatible, it can't know with which field to merge.
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, builder.add_field(f0_opt));

        AssertSchemaBuilderYield(builder, schema({f0, f0}));
    }

    TEST(TestSchemaBuilder, PolicyError) {
        auto f0 = field("f0", int32(), true);
        auto f1 = field("f1", uint8());
        // Same as f0, but not required.
        auto f0_opt = field("f0", int32());
        // Another type, can't merge
        auto f0_other = field("f0", utf8(), false);

        SchemaBuilder builder{SchemaBuilder::CONFLICT_ERROR};

        ASSERT_OK(builder.add_fields({f0, f1}));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));

        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, builder.add_field(f0));
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, builder.add_field(f0_opt));
        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, builder.add_field(f0_other));
        AssertSchemaBuilderYield(builder, schema({f0, f1}));
    }

    TEST(TestSchemaBuilder, Merge) {
        auto f0 = field("f0", int32(), true);
        auto f1 = field("f1", uint8());
        // Same as f0, but not required.
        auto f0_opt = field("f0", int32());
        // Another type, can't merge
        auto f0_other = field("f0", utf8(), false);

        auto s1 = schema({f0, f1});
        auto s2 = schema({f1, f0});
        auto s3 = schema({f0_opt});
        auto broken = schema({f0_other});

        ASSERT_OK_AND_ASSIGN(auto schema, SchemaBuilder::Merge({s1, s2, s3}));
        ASSERT_OK(SchemaBuilder::are_compatible({s1, s2, s3}));
        AssertSchemaEqual(schema, ::nebula::schema({f0_opt, f1}));

        ASSERT_OK_AND_ASSIGN(schema, SchemaBuilder::Merge({s2, s3, s1}));
        AssertSchemaEqual(schema, ::nebula::schema({f1, f0_opt}));

        ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, SchemaBuilder::Merge({s3, broken}));
        ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, SchemaBuilder::are_compatible({s3, broken}));
    }

    class TestUnifySchemas : public TestSchema {
    protected:
        void AssertSchemaEqualsUnorderedFields(const Schema &lhs, const Schema &rhs) {
            if (lhs.metadata()) {
                ASSERT_NE(nullptr, rhs.metadata());
                ASSERT_TRUE(lhs.metadata()->equals(*rhs.metadata()));
            } else {
                ASSERT_EQ(nullptr, rhs.metadata());
            }
            ASSERT_EQ(lhs.num_fields(), rhs.num_fields());
            for (int i = 0; i < lhs.num_fields(); ++i) {
                auto lhs_field = lhs.field(i);
                auto rhs_field = rhs.get_field_by_name(lhs_field->name());
                ASSERT_NE(nullptr, rhs_field);
                ASSERT_TRUE(lhs_field->equals(rhs_field, true))
          << lhs_field->to_string() << " vs " << rhs_field->to_string();
            }
        }

        void CheckUnifyAsymmetric(
            const std::shared_ptr<Field> &field1, const std::shared_ptr<Field> &field2,
            const std::shared_ptr<Field> &expected,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            NEBULA_SCOPED_TRACE("options: ", options);
            NEBULA_SCOPED_TRACE("field2: ", field2->to_string());
            NEBULA_SCOPED_TRACE("field1: ", field1->to_string());
            ASSERT_OK_AND_ASSIGN(auto merged, field1->merge_with(field2, options));
            AssertFieldEqual(merged, expected);
        }

        void CheckPromoteTo(
            const std::shared_ptr<Field> &field1, const std::shared_ptr<Field> &field2,
            const std::shared_ptr<Field> &expected,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            CheckUnifyAsymmetric(field1, field2, expected, options);
            CheckUnifyAsymmetric(field2, field1, expected, options);
        }

        void CheckUnifyFailsInvalid(
            const std::shared_ptr<Field> &field1, const std::shared_ptr<Field> &field2,
            const Field::MergeOptions &options = Field::MergeOptions::defaults(),
            const std::string &match_message = "") {
            NEBULA_SCOPED_TRACE("options: ", options);
            NEBULA_SCOPED_TRACE("field2: ", field2->to_string());
            NEBULA_SCOPED_TRACE("field1: ", field1->to_string());
            EXPECT_RAISES_WITH_MESSAGE_THAT(turbo::StatusCode::kInvalidArgument, ::testing::HasSubstr(match_message),
                                            field1->merge_with(field2, options));
            EXPECT_RAISES_WITH_MESSAGE_THAT(turbo::StatusCode::kInvalidArgument, ::testing::HasSubstr(match_message),
                                            field2->merge_with(field1, options));
        }

        void CheckUnifyFailsTypeError(
            const std::shared_ptr<Field> &field1, const std::shared_ptr<Field> &field2,
            const Field::MergeOptions &options = Field::MergeOptions::defaults(),
            const std::string &match_message = "") {
            NEBULA_SCOPED_TRACE("options: ", options);
            NEBULA_SCOPED_TRACE("field2: ", field2->to_string());
            NEBULA_SCOPED_TRACE("field1: ", field1->to_string());
            ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, field1->merge_with(field2, options));
            ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, field2->merge_with(field1, options));
            EXPECT_RAISES_WITH_MESSAGE_THAT(turbo::StatusCode::kFailedPrecondition, ::testing::HasSubstr(match_message),
                                            field1->merge_with(field2, options));
            EXPECT_RAISES_WITH_MESSAGE_THAT(turbo::StatusCode::kFailedPrecondition, ::testing::HasSubstr(match_message),
                                            field2->merge_with(field1, options));
        }

        void CheckPromoteTo(
            const std::shared_ptr<DataType> &left, const std::shared_ptr<DataType> &right,
            const std::shared_ptr<DataType> &expected,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            auto field1 = field("a", left);
            auto field2 = field("a", right);
            CheckPromoteTo(field1, field2, field("a", expected), options);

            field1 = field("a", left, /*nullable=*/false);
            field2 = field("a", right, /*nullable=*/false);
            CheckPromoteTo(field1, field2, field("a", expected, /*nullable=*/false), options);

            field1 = field("a", left);
            field2 = field("a", right, /*nullable=*/false);
            CheckPromoteTo(field1, field2, field("a", expected, /*nullable=*/true), options);

            field1 = field("a", left, /*nullable=*/false);
            field2 = field("a", right);
            CheckPromoteTo(field1, field2, field("a", expected, /*nullable=*/true), options);
        }

        void CheckUnifyAsymmetric(
            const std::shared_ptr<DataType> &left, const std::shared_ptr<DataType> &right,
            const std::shared_ptr<DataType> &expected,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            auto field1 = field("a", left);
            auto field2 = field("a", right);
            CheckUnifyAsymmetric(field1, field2, field("a", expected), options);

            field1 = field("a", left, /*nullable=*/false);
            field2 = field("a", right, /*nullable=*/false);
            CheckUnifyAsymmetric(field1, field2, field("a", expected, /*nullable=*/false),
                                 options);

            field1 = field("a", left);
            field2 = field("a", right, /*nullable=*/false);
            CheckUnifyAsymmetric(field1, field2, field("a", expected, /*nullable=*/true),
                                 options);

            field1 = field("a", left, /*nullable=*/false);
            field2 = field("a", right);
            CheckUnifyAsymmetric(field1, field2, field("a", expected, /*nullable=*/true),
                                 options);
        }

        void CheckPromoteTo(
            const std::shared_ptr<DataType> &from,
            const std::vector<std::shared_ptr<DataType> > &to,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            for (const auto &ty: to) {
                CheckPromoteTo(from, ty, ty, options);
            }
        }

        void CheckUnifyFailsInvalid(
            const std::shared_ptr<DataType> &left, const std::shared_ptr<DataType> &right,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            auto field1 = field("a", left);
            auto field2 = field("a", right);
            CheckUnifyFailsInvalid(field1, field2, options);
        }

        void CheckUnifyFailsInvalid(
            const std::shared_ptr<DataType> &from,
            const std::vector<std::shared_ptr<DataType> > &to,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            for (const auto &ty: to) {
                CheckUnifyFailsInvalid(from, ty, options);
            }
        }

        void CheckUnifyFailsInvalid(
            const std::vector<std::shared_ptr<DataType> > &from,
            const std::vector<std::shared_ptr<DataType> > &to,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            for (const auto &ty: from) {
                CheckUnifyFailsInvalid(ty, to, options);
            }
        }

        void CheckUnifyFailsTypeError(
            const std::shared_ptr<DataType> &left, const std::shared_ptr<DataType> &right,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            auto field1 = field("a", left);
            auto field2 = field("a", right);
            CheckUnifyFailsTypeError(field1, field2, options);
        }

        void CheckUnifyFailsTypeError(
            const std::shared_ptr<DataType> &from,
            const std::vector<std::shared_ptr<DataType> > &to,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            for (const auto &ty: to) {
                CheckUnifyFailsTypeError(from, ty, options);
            }
        }

        void CheckUnifyFailsTypeError(
            const std::vector<std::shared_ptr<DataType> > &from,
            const std::vector<std::shared_ptr<DataType> > &to,
            const Field::MergeOptions &options = Field::MergeOptions::defaults()) {
            for (const auto &ty: from) {
                CheckUnifyFailsTypeError(ty, to, options);
            }
        }
    };

    TEST_F(TestUnifySchemas, EmptyInput) { ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, unify_schemas({})); }

    TEST_F(TestUnifySchemas, IdenticalSchemas) {
        auto int32_field = field("int32_field", int32());
        auto uint8_field = field("uint8_field", uint8(), false);
        auto utf8_field = field("utf8_field", utf8());
        std::vector<std::string> keys{"foo"};
        std::vector<std::string> vals{"bar"};
        auto metadata = std::make_shared<KeyValueMetadata>(keys, vals);

        auto schema1 = schema({int32_field, uint8_field, utf8_field});
        auto schema2 = schema({int32_field, uint8_field, utf8_field->with_metadata(metadata)})
                ->with_metadata(metadata);

        ASSERT_OK_AND_ASSIGN(auto result, unify_schemas({schema1, schema2}));
        // Using Schema::Equals to make sure the ordering of fields is not changed.
        ASSERT_TRUE(result->equals(*schema1, /*check_metadata=*/true));

        ASSERT_OK_AND_ASSIGN(result, unify_schemas({schema2, schema1}));
        // Using Schema::Equals to make sure the ordering of fields is not changed.
        ASSERT_TRUE(result->equals(*schema2, /*check_metadata=*/true));
    }

    TEST_F(TestUnifySchemas, FieldOrderingSameAsTheFirstSchema) {
        auto int32_field = field("int32_field", int32());
        auto uint8_field = field("uint8_field", uint8(), false);
        auto utf8_field = field("utf8_field", utf8());
        auto binary_field = field("binary_field", binary());

        auto schema1 = schema({int32_field, uint8_field, utf8_field});
        // schema2 only differs from schema1 in field ordering.
        auto schema2 = schema({uint8_field, int32_field, utf8_field});
        auto schema3 = schema({binary_field});

        ASSERT_OK_AND_ASSIGN(auto result, unify_schemas({schema1, schema2, schema3}));

        ASSERT_EQ(4, result->num_fields());
        ASSERT_TRUE(int32_field->equals(result->field(0)));
        ASSERT_TRUE(uint8_field->equals(result->field(1)));
        ASSERT_TRUE(utf8_field->equals(result->field(2)));
        ASSERT_TRUE(binary_field->equals(result->field(3)));
    }

    TEST_F(TestUnifySchemas, MissingField) {
        auto int32_field = field("int32_field", int32());
        auto uint8_field = field("uint8_field", uint8(), false);
        auto utf8_field = field("utf8_field", utf8());
        auto metadata1 = key_value_metadata({"foo"}, {"bar"});
        auto metadata2 = key_value_metadata({"q"}, {"42"});

        auto schema1 = schema({int32_field, uint8_field})->with_metadata(metadata1);
        auto schema2 = schema({uint8_field, utf8_field->with_metadata(metadata2)});
        auto schema3 = schema({int32_field->with_metadata(metadata1), uint8_field, utf8_field});

        ASSERT_OK_AND_ASSIGN(auto result, unify_schemas({schema1, schema2}));
        AssertSchemaEqualsUnorderedFields(
            *result, *schema({int32_field, uint8_field, utf8_field->with_metadata(metadata2)})
            ->with_metadata(metadata1));
    }

    TEST_F(TestUnifySchemas, PromoteNullTypeField) {
        auto metadata = key_value_metadata({"foo"}, {"bar"});
        auto null_field = field("f", null());
        auto int32_field = field("f", int32(), /*nullable=*/false);

        auto schema1 = schema({null_field->with_metadata(metadata)});
        auto schema2 = schema({int32_field});

        ASSERT_OK_AND_ASSIGN(auto result, unify_schemas({schema1, schema2}));
        AssertSchemaEqualsUnorderedFields(
            *result, *schema({int32_field->with_metadata(metadata)->with_nullable(true)}));

        ASSERT_OK_AND_ASSIGN(result, unify_schemas({schema2, schema1}));
        AssertSchemaEqualsUnorderedFields(*result, *schema({int32_field->with_nullable(true)}));
    }

    TEST_F(TestUnifySchemas, MoreSchemas) {
        auto int32_field = field("int32_field", int32());
        auto uint8_field = field("uint8_field", uint8(), false);
        auto utf8_field = field("utf8_field", utf8());

        ASSERT_OK_AND_ASSIGN(
            auto result,
            unify_schemas({schema({int32_field}), schema({uint8_field}), schema({utf8_field})}));
        AssertSchemaEqualsUnorderedFields(
            *result, *schema({
                int32_field->with_nullable(true), uint8_field->with_nullable(false),
                utf8_field->with_nullable(true)
            }));
    }

    TEST_F(TestUnifySchemas, Numeric) {
        auto options = Field::MergeOptions::defaults();
        options.promote_numeric_width = true;
        options.promote_integer_to_float = true;
        options.promote_integer_sign = true;
        CheckPromoteTo(
            uint8(),
            {uint16(), int16(), uint32(), int32(), uint64(), int64(), float32(), float64()},
            options);
        CheckPromoteTo(int8(), {int16(), int32(), int64(), float32(), float64()}, options);
        CheckPromoteTo(uint16(), {uint32(), int32(), uint64(), int64(), float32(), float64()},
                       options);
        CheckPromoteTo(int16(), {int32(), int64(), float32(), float64()}, options);
        CheckPromoteTo(uint32(), {uint64(), int64(), float64()}, options);
        CheckPromoteTo(int32(), {int64(), float64()}, options);
        CheckPromoteTo(uint64(), {int64(), float64()}, options);
        CheckPromoteTo(int64(), {float64()}, options);
        CheckPromoteTo(float16(), {float32(), float64()}, options);
        CheckPromoteTo(float32(), {float64()}, options);
        CheckPromoteTo(uint64(), float32(), float64(), options);
        CheckPromoteTo(int64(), float32(), float64(), options);

        options.promote_integer_sign = false;
        CheckPromoteTo(uint8(), {uint16(), uint32(), uint64()}, options);
        CheckPromoteTo(int8(), {int16(), int32(), int64()}, options);
        CheckUnifyFailsTypeError(uint8(), {int8(), int16(), int32(), int64()}, options);
        CheckPromoteTo(uint16(), {uint32(), uint64()}, options);
        CheckPromoteTo(int16(), {int32(), int64()}, options);
        CheckUnifyFailsTypeError(uint16(), {int16(), int32(), int64()}, options);
        CheckPromoteTo(uint32(), {uint64()}, options);
        CheckPromoteTo(int32(), {int64()}, options);
        CheckUnifyFailsTypeError(uint32(), {int32(), int64()}, options);
        CheckUnifyFailsTypeError(uint64(), {int64()}, options);

        options.promote_integer_sign = true;
        options.promote_integer_to_float = false;
        CheckUnifyFailsTypeError(int_types(), floating_point_types(), options);

        options.promote_integer_to_float = true;
        options.promote_numeric_width = false;
        CheckUnifyFailsTypeError(int8(), {int16(), int32(), int64()}, options);
        CheckUnifyFailsTypeError(int16(), {int32(), int64()}, options);
        CheckUnifyFailsTypeError(int32(), {int64()}, options);
        CheckUnifyFailsTypeError(int32(), {float16(), float32()}, options);
        CheckPromoteTo(int32(), {float64()}, options);
        CheckPromoteTo(int64(), {float64()}, options);

        CheckPromoteTo(uint8(), int8(), int16(), options);
        CheckPromoteTo(uint16(), int8(), int32(), options);
        CheckPromoteTo(uint32(), int8(), int64(), options);
        CheckPromoteTo(uint32(), int32(), int64(), options);
    }

    TEST_F(TestUnifySchemas, Decimal) {
        auto options = Field::MergeOptions::defaults();

        options.promote_decimal_to_float = true;
        CheckPromoteTo(decimal128(3, 2), {float32(), float64()}, options);
        CheckPromoteTo(decimal256(3, 2), {float32(), float64()}, options);

        options.promote_integer_to_decimal = true;
        CheckPromoteTo(int32(), decimal128(3, 2), decimal128(12, 2), options);
        CheckPromoteTo(int32(), decimal128(3, -2), decimal128(10, 0), options);

        options.promote_decimal = true;
        CheckPromoteTo(decimal128(3, 2), decimal128(5, 2), decimal128(5, 2), options);
        CheckPromoteTo(decimal128(3, 2), decimal128(5, 3), decimal128(5, 3), options);
        CheckPromoteTo(decimal128(3, 2), decimal128(5, 1), decimal128(6, 2), options);
        CheckPromoteTo(decimal128(3, 2), decimal128(5, -2), decimal128(9, 2), options);
        CheckPromoteTo(decimal128(3, -2), decimal128(5, -2), decimal128(5, -2), options);
        CheckPromoteTo(decimal128(38, 10), decimal128(38, 5), decimal256(43, 10), options);

        CheckPromoteTo(decimal256(3, 2), decimal256(5, 2), decimal256(5, 2), options);
        CheckPromoteTo(decimal256(3, 2), decimal256(5, 3), decimal256(5, 3), options);
        CheckPromoteTo(decimal256(3, 2), decimal256(5, 1), decimal256(6, 2), options);
        CheckPromoteTo(decimal256(3, 2), decimal256(5, -2), decimal256(9, 2), options);
        CheckPromoteTo(decimal256(3, -2), decimal256(5, -2), decimal256(5, -2), options);

        // int32() is essentially decimal128(10, 0)
        CheckPromoteTo(int32(), decimal128(3, 2), decimal128(12, 2), options);
        CheckPromoteTo(int32(), decimal128(3, -2), decimal128(10, 0), options);
        CheckPromoteTo(int64(), decimal128(38, 37), decimal256(56, 37), options);

        CheckUnifyFailsTypeError(decimal256(1, 0), decimal128(1, 0), options);

        options.promote_numeric_width = true;
        CheckPromoteTo(decimal128(3, 2), decimal256(5, 2), decimal256(5, 2), options);
        CheckPromoteTo(int32(), decimal128(38, 37), decimal256(47, 37), options);
        CheckUnifyFailsInvalid(decimal128(38, 10), decimal256(76, 5), options);

        CheckUnifyFailsInvalid(int64(), decimal256(76, 75), options);
    }

    TEST_F(TestUnifySchemas, Temporal) {
        auto options = Field::MergeOptions::defaults();

        options.promote_temporal_unit = true;
        CheckPromoteTo(date32(), {date64()}, options);

        CheckPromoteTo(
            time32(TimeUnit::SECOND),
            {time32(TimeUnit::MILLI), time64(TimeUnit::MICRO), time64(TimeUnit::NANO)},
            options);
        CheckPromoteTo(time32(TimeUnit::MILLI),
                       {time64(TimeUnit::MICRO), time64(TimeUnit::NANO)}, options);
        CheckPromoteTo(time64(TimeUnit::MICRO), {time64(TimeUnit::NANO)}, options);

        CheckPromoteTo(
            duration(TimeUnit::SECOND),
            {duration(TimeUnit::MILLI), duration(TimeUnit::MICRO), duration(TimeUnit::NANO)},
            options);
        CheckPromoteTo(duration(TimeUnit::MILLI),
                       {duration(TimeUnit::MICRO), duration(TimeUnit::NANO)}, options);
        CheckPromoteTo(duration(TimeUnit::MICRO), {duration(TimeUnit::NANO)}, options);

        CheckPromoteTo(
            timestamp(TimeUnit::SECOND),
            {timestamp(TimeUnit::MILLI), timestamp(TimeUnit::MICRO), timestamp(TimeUnit::NANO)},
            options);
        CheckPromoteTo(timestamp(TimeUnit::MILLI),
                       {timestamp(TimeUnit::MICRO), timestamp(TimeUnit::NANO)}, options);
        CheckPromoteTo(timestamp(TimeUnit::MICRO), {timestamp(TimeUnit::NANO)}, options);

        CheckUnifyFailsTypeError(timestamp(TimeUnit::SECOND),
                                 timestamp(TimeUnit::SECOND, "UTC"), options);
        CheckUnifyFailsTypeError(timestamp(TimeUnit::SECOND, "America/New_York"),
                                 timestamp(TimeUnit::SECOND, "UTC"), options);

        options.promote_temporal_unit = false;
        CheckUnifyFailsTypeError(timestamp(TimeUnit::MICRO), timestamp(TimeUnit::NANO),
                                 options);
    }

    TEST_F(TestUnifySchemas, Binary) {
        auto options = Field::MergeOptions::defaults();
        options.promote_binary = true;
        CheckPromoteTo(utf8(), {large_utf8(), binary(), large_binary()}, options);
        CheckPromoteTo(binary(), {large_binary()}, options);
        CheckPromoteTo(fixed_size_binary(2), {fixed_size_binary(2), binary(), large_binary()},
                       options);
        CheckPromoteTo(fixed_size_binary(2), fixed_size_binary(4), binary(), options);

        options.promote_binary = false;
        CheckUnifyFailsTypeError({utf8(), binary()}, {large_utf8(), large_binary()});
        CheckUnifyFailsTypeError(fixed_size_binary(2), base_binary_types());
        CheckUnifyFailsTypeError(utf8(), {binary(), large_binary(), fixed_size_binary(2)});
    }

    TEST_F(TestUnifySchemas, List) {
        auto options = Field::MergeOptions::defaults();
        options.promote_list = true;

        CheckPromoteTo(fixed_size_list(int8(), 2), fixed_size_list(int8(), 3), list(int8()));

        CheckPromoteTo(list(int8()), {large_list(int8())}, options);
        CheckPromoteTo(fixed_size_list(int8(), 2), {list(int8()), large_list(int8())}, options);

        options.promote_numeric_width = true;
        CheckPromoteTo(list(int8()), {list(int16()), list(int32()), list(int64())}, options);
        CheckPromoteTo(
            fixed_size_list(int8(), 2),
            {fixed_size_list(int16(), 2), list(int16()), list(int32()), list(int64())},
            options);
        CheckPromoteTo(fixed_size_list(int16(), 2), list(int8()), list(int16()), options);

        auto ty = list(field("foo", int8(), /*nullable=*/false));
        CheckUnifyAsymmetric(ty, list(int8()), list(field("foo", int8(), /*nullable=*/true)),
                             options);
        CheckUnifyAsymmetric(ty, list(field("bar", int16(), /*nullable=*/false)),
                             list(field("foo", int16(), /*nullable=*/false)), options);

        options.promote_list = false;
        CheckUnifyFailsTypeError(list(int8()), large_list(int8()));
    }

    TEST_F(TestUnifySchemas, Map) {
        auto options = Field::MergeOptions::defaults();
        options.promote_numeric_width = true;

        CheckPromoteTo(map(int8(), int32()),
                       {map(int8(), int64()), map(int16(), int32()), map(int64(), int64())},
                       options);

        // Do not test field names, since MapType intentionally ignores them in comparisons
        // See ARROW-7173, ARROW-14999
        auto ty = map(int8(), field("value", int32(), /*nullable=*/false));
        CheckPromoteTo(ty, map(int8(), int32()),
                       map(int8(), field("value", int32(), /*nullable=*/true)), options);
        CheckPromoteTo(ty, map(int16(), field("value", int64(), /*nullable=*/false)),
                       map(int16(), field("value", int64(), /*nullable=*/false)), options);
    }

    TEST_F(TestUnifySchemas, Struct) {
        auto options = Field::MergeOptions::defaults();
        options.promote_numeric_width = true;
        options.promote_binary = true;

        CheckPromoteTo(STRUCT({}), STRUCT({field("a", int8())}),
                       STRUCT({field("a", int8())}), options);

        CheckUnifyAsymmetric(STRUCT({field("b", utf8())}), STRUCT({field("a", int8())}),
                             STRUCT({field("b", utf8()), field("a", int8())}), options);
        CheckUnifyAsymmetric(STRUCT({field("a", int8())}), STRUCT({field("b", utf8())}),
                             STRUCT({field("a", int8()), field("b", utf8())}), options);

        CheckPromoteTo(STRUCT({field("b", utf8())}), STRUCT({field("b", binary())}),
                       STRUCT({field("b", binary())}), options);

        CheckUnifyAsymmetric(
            STRUCT({field("a", int8()), field("b", utf8()), field("a", int64())}),
            STRUCT({field("b", binary())}),
            STRUCT({field("a", int8()), field("b", binary()), field("a", int64())}), options);

        ASSERT_RAISES(
            turbo::StatusCode::kInvalidArgument,
            field("foo", STRUCT({field("a", int8()), field("b", utf8()), field("a", int64())}))
            ->merge_with(field("foo", STRUCT({field("a", int64())})), options));
    }

    TEST_F(TestUnifySchemas, Dictionary) {
        auto options = Field::MergeOptions::defaults();
        options.promote_dictionary = true;
        options.promote_binary = true;

        CheckPromoteTo(dictionary(int8(), utf8()),
                       {
                           dictionary(int64(), utf8()),
                           dictionary(int8(), large_utf8()),
                       },
                       options);
        CheckPromoteTo(dictionary(int64(), utf8()), dictionary(int8(), large_utf8()),
                       dictionary(int64(), large_utf8()), options);
        CheckPromoteTo(dictionary(int8(), utf8(), /*ordered=*/true),
                       {
                           dictionary(int64(), utf8(), /*ordered=*/true),
                           dictionary(int8(), large_utf8(), /*ordered=*/true),
                       },
                       options);
        CheckUnifyFailsTypeError(dictionary(int8(), utf8()),
                                 dictionary(int8(), utf8(), /*ordered=*/true), options);

        options.promote_dictionary_ordered = true;
        CheckPromoteTo(dictionary(int8(), utf8()), dictionary(int8(), utf8(), /*ordered=*/true),
                       dictionary(int8(), utf8(), /*ordered=*/false), options);
    }

    TEST_F(TestUnifySchemas, IncompatibleTypes) {
        auto int32_field = field("f", int32());
        auto uint8_field = field("f", uint8(), false);

        auto schema1 = schema({int32_field});
        auto schema2 = schema({uint8_field});

        ASSERT_RAISES(turbo::StatusCode::kFailedPrecondition, unify_schemas({schema1, schema2}));
    }

    TEST_F(TestUnifySchemas, DuplicateFieldNames) {
        auto int32_field = field("int32_field", int32());
        auto utf8_field = field("utf8_field", utf8());

        auto schema1 = schema({int32_field, utf8_field});
        auto schema2 = schema({int32_field, int32_field, utf8_field});

        ASSERT_RAISES(turbo::StatusCode::kInvalidArgument, unify_schemas({schema1, schema2}));
    }

#define PRIMITIVE_TEST(KLASS, CTYPE, ENUM, NAME)                              \
  TEST(TypesTest, TURBO_CONCAT(TestPrimitive_, ENUM)) {                       \
    KLASS tp;                                                                 \
                                                                              \
    ASSERT_EQ(tp.id(), Type::ENUM);                                           \
    ASSERT_EQ(tp.to_string(), std::string(NAME));                              \
                                                                              \
    using CType = TypeTraits<KLASS>::CType;                                   \
    static_assert(std::is_same<CType, CTYPE>::value, "Not the same c-type!"); \
                                                                              \
    using DerivedArrowType = CTypeTraits<CTYPE>::ArrowType;                   \
    static_assert(std::is_same<DerivedArrowType, KLASS>::value,               \
                  "Not the same nebula-type!");                                \
  }

    PRIMITIVE_TEST(Int8Type, int8_t, INT8, "int8");
    PRIMITIVE_TEST(Int16Type, int16_t, INT16, "int16");
    PRIMITIVE_TEST(Int32Type, int32_t, INT32, "int32");
    PRIMITIVE_TEST(Int64Type, int64_t, INT64, "int64");
    PRIMITIVE_TEST(UInt8Type, uint8_t, UINT8, "uint8");
    PRIMITIVE_TEST(UInt16Type, uint16_t, UINT16, "uint16");
    PRIMITIVE_TEST(UInt32Type, uint32_t, UINT32, "uint32");
    PRIMITIVE_TEST(UInt64Type, uint64_t, UINT64, "uint64");

    PRIMITIVE_TEST(Fp32Type, float, FP32, "float");
    PRIMITIVE_TEST(Fp64Type, double, FP64, "double");

    PRIMITIVE_TEST(BooleanType, bool, BOOL, "bool");

    TEST(TestBinaryType, ToString) {
        BinaryType t1;
        BinaryType e1;
        StringType t2;
        AssertTypeEqual(t1, e1);
        AssertTypeNotEqual(t1, t2);
        ASSERT_EQ(t1.id(), Type::BINARY);
        ASSERT_EQ(t1.to_string(), std::string("binary"));
    }

    TEST(TestStringType, ToString) {
        StringType str;
        ASSERT_EQ(str.id(), Type::STRING);
        ASSERT_EQ(str.name(), std::string("utf8"));
        ASSERT_EQ(str.type_name(), std::string("utf8"));
        ASSERT_EQ(str.to_string(), std::string("string"));
    }

    TEST(TestBinaryViewType, ToString) {
        BinaryViewType t1;
        BinaryViewType e1;
        StringViewType t2;
        AssertTypeEqual(t1, e1);
        AssertTypeNotEqual(t1, t2);
        ASSERT_EQ(t1.id(), Type::BINARY_VIEW);
        ASSERT_EQ(t1.to_string(), std::string("binary_view"));
    }

    TEST(TestLargeBinaryTypes, ToString) {
        BinaryType bt1;
        LargeBinaryType t1;
        LargeBinaryType e1;
        LargeStringType t2;
        AssertTypeEqual(t1, e1);
        AssertTypeNotEqual(t1, t2);
        AssertTypeNotEqual(t1, bt1);
        ASSERT_EQ(t1.id(), Type::LARGE_BINARY);
        ASSERT_EQ(t1.to_string(), std::string("large_binary"));
        ASSERT_EQ(t2.id(), Type::LARGE_STRING);
        ASSERT_EQ(t2.to_string(), std::string("large_string"));
    }

    TEST(TestFixedSizeBinaryType, ToString) {
        auto t = fixed_size_binary(10);
        ASSERT_EQ(t->id(), Type::FIXED_SIZE_BINARY);
        ASSERT_EQ("fixed_size_binary[10]", t->to_string());
    }

    TEST(TestFixedSizeBinaryType, Equals) {
        auto t1 = fixed_size_binary(10);
        auto t2 = fixed_size_binary(10);
        auto t3 = fixed_size_binary(3);

        AssertTypeEqual(*t1, *t2);
        AssertTypeNotEqual(*t1, *t3);
    }

    TEST(TestListType, Basics) {
        std::shared_ptr<DataType> vt = std::make_shared<UInt8Type>();

        ListType list_type(vt);
        ASSERT_EQ(list_type.id(), Type::LIST);

        ASSERT_EQ("list", list_type.name());
        ASSERT_EQ("list<item: uint8>", list_type.to_string());

        ASSERT_EQ(list_type.get_value_type()->id(), vt->id());
        ASSERT_EQ(list_type.get_value_type()->id(), vt->id());

        std::shared_ptr<DataType> st = std::make_shared<StringType>();
        std::shared_ptr<DataType> lt = std::make_shared<ListType>(st);
        ASSERT_EQ("list<item: string>", lt->to_string());

        ListType lt2(lt);
        ASSERT_EQ("list<item: list<item: string>>", lt2.to_string());
    }

    TEST(TestLargeListType, Basics) {
        std::shared_ptr<DataType> vt = std::make_shared<UInt8Type>();

        LargeListType list_type(vt);
        ASSERT_EQ(list_type.id(), Type::LARGE_LIST);

        ASSERT_EQ("large_list", list_type.name());
        ASSERT_EQ("large_list<item: uint8>", list_type.to_string());

        ASSERT_EQ(list_type.get_value_type()->id(), vt->id());
        ASSERT_EQ(list_type.get_value_type()->id(), vt->id());

        std::shared_ptr<DataType> st = std::make_shared<StringType>();
        std::shared_ptr<DataType> lt = std::make_shared<LargeListType>(st);
        ASSERT_EQ("large_list<item: string>", lt->to_string());

        LargeListType lt2(lt);
        ASSERT_EQ("large_list<item: large_list<item: string>>", lt2.to_string());
    }

    TEST(TestListViewType, Basics) {
        std::shared_ptr<DataType> vt = std::make_shared<UInt8Type>();

        ListViewType list_view_type(vt);
        ASSERT_EQ(list_view_type.id(), Type::LIST_VIEW);

        ASSERT_EQ("list_view", list_view_type.name());
        ASSERT_EQ("list_view<item: uint8>", list_view_type.to_string());

        ASSERT_EQ(list_view_type.get_value_type()->id(), vt->id());
        ASSERT_EQ(list_view_type.get_value_type()->id(), vt->id());

        std::shared_ptr<DataType> st = std::make_shared<StringType>();
        std::shared_ptr<DataType> lt = std::make_shared<ListViewType>(st);
        ASSERT_EQ("list_view<item: string>", lt->to_string());

        ListViewType lt2(lt);
        ASSERT_EQ("list_view<item: list_view<item: string>>", lt2.to_string());
    }

    TEST(TestLargeListViewType, Basics) {
        std::shared_ptr<DataType> vt = std::make_shared<UInt8Type>();

        LargeListViewType list_view_type(vt);
        ASSERT_EQ(list_view_type.id(), Type::LARGE_LIST_VIEW);

        ASSERT_EQ("large_list_view", list_view_type.name());
        ASSERT_EQ("large_list_view<item: uint8>", list_view_type.to_string());

        ASSERT_EQ(list_view_type.get_value_type()->id(), vt->id());
        ASSERT_EQ(list_view_type.get_value_type()->id(), vt->id());

        std::shared_ptr<DataType> st = std::make_shared<StringType>();
        std::shared_ptr<DataType> lt = std::make_shared<LargeListViewType>(st);
        ASSERT_EQ("large_list_view<item: string>", lt->to_string());

        LargeListViewType lt2(lt);
        ASSERT_EQ("large_list_view<item: large_list_view<item: string>>", lt2.to_string());
    }

    TEST(TestMapType, Basics) {
        auto md = key_value_metadata({"foo"}, {"foo value"});

        std::shared_ptr<DataType> kt = std::make_shared<StringType>();
        std::shared_ptr<DataType> it = std::make_shared<UInt8Type>();

        MapType map_type(kt, it);
        ASSERT_EQ(map_type.id(), Type::MAP);

        ASSERT_EQ("map", map_type.name());
        ASSERT_EQ("map<string, uint8>", map_type.to_string());

        ASSERT_EQ(map_type.key_type()->id(), kt->id());
        ASSERT_EQ(map_type.item_type()->id(), it->id());
        ASSERT_EQ(map_type.get_value_type()->id(), Type::STRUCT);

        std::shared_ptr<DataType> mt = std::make_shared<MapType>(it, kt);
        ASSERT_EQ("map<uint8, string>", mt->to_string());

        MapType mt2(kt, mt, /*keys_sorted=*/true);
        ASSERT_EQ("map<string, map<uint8, string>, keys_sorted>", mt2.to_string());
        AssertTypeNotEqual(map_type, mt2);
        MapType mt3(kt, mt);
        ASSERT_EQ("map<string, map<uint8, string>>", mt3.to_string());
        AssertTypeNotEqual(mt2, mt3);
        MapType mt4(kt, mt);
        AssertTypeEqual(mt3, mt4);

        // Field names are indifferent when comparing map types
        ASSERT_OK_AND_ASSIGN(
            auto mt5,
            MapType::create(field(
                "some_entries",
                STRUCT({field("some_key", kt, false), field("some_value", mt)}), false)));
        AssertTypeEqual(mt3, *mt5);
        // ...unless we explicitly ask about them.
        ASSERT_FALSE(mt3.equals(mt5, /*check_metadata=*/true));

        // nullability of value type matters in comparisons
        MapType map_type_non_nullable(kt, field("value", it, /*nullable=*/false));
        AssertTypeNotEqual(map_type, map_type_non_nullable);
    }

    TEST(TestMapType, Metadata) {
        auto md1 = key_value_metadata({"foo", "bar"}, {"foo value", "bar value"});
        auto md2 = key_value_metadata({"foo", "bar"}, {"foo value", "bar value"});
        auto md3 = key_value_metadata({"foo"}, {"foo value"});

        auto t1 = map(utf8(), field("value", int32(), md1));
        auto t2 = map(utf8(), field("value", int32(), md2));
        auto t3 = map(utf8(), field("value", int32(), md3));
        auto t4 =
                std::make_shared<MapType>(field("key", utf8(), md1), field("value", int32(), md2));
        ASSERT_OK_AND_ASSIGN(auto t5,
                             MapType::create(field("some_entries",
                                 STRUCT({field("some_key", utf8(), false),
                                     field("some_value", int32(), md2)}),
                                 false, md2)));

        AssertTypeEqual(*t1, *t2);
        AssertTypeEqual(*t1, *t2, /*check_metadata=*/true);

        AssertTypeEqual(*t1, *t3);
        AssertTypeNotEqual(*t1, *t3, /*check_metadata=*/true);

        AssertTypeEqual(*t1, *t4);
        AssertTypeNotEqual(*t1, *t4, /*check_metadata=*/true);

        AssertTypeEqual(*t1, *t5);
        AssertTypeNotEqual(*t1, *t5, /*check_metadata=*/true);
    }

    TEST(TestFixedSizeListType, Basics) {
        std::shared_ptr<DataType> vt = std::make_shared<UInt8Type>();

        FixedSizeListType fixed_size_list_type(vt, 4);
        ASSERT_EQ(fixed_size_list_type.id(), Type::FIXED_SIZE_LIST);

        ASSERT_EQ(4, fixed_size_list_type.list_size());
        ASSERT_EQ("fixed_size_list", fixed_size_list_type.name());
        ASSERT_EQ("fixed_size_list<item: uint8>[4]", fixed_size_list_type.to_string());

        ASSERT_EQ(fixed_size_list_type.get_value_type()->id(), vt->id());
        ASSERT_EQ(fixed_size_list_type.get_value_type()->id(), vt->id());

        std::shared_ptr<DataType> st = std::make_shared<StringType>();
        std::shared_ptr<DataType> lt = std::make_shared<FixedSizeListType>(st, 3);
        ASSERT_EQ("fixed_size_list<item: string>[3]", lt->to_string());

        FixedSizeListType lt2(lt, 7);
        ASSERT_EQ("fixed_size_list<item: fixed_size_list<item: string>[3]>[7]", lt2.to_string());
    }

    TEST(TestFixedSizeListType, Equals) {
        auto t1 = fixed_size_list(int8(), 3);
        auto t2 = fixed_size_list(int8(), 3);
        auto t3 = fixed_size_list(int8(), 4);
        auto t4 = fixed_size_list(int16(), 4);
        auto t5 = fixed_size_list(list(int16()), 4);
        auto t6 = fixed_size_list(list(int16()), 4);
        auto t7 = fixed_size_list(list(int32()), 4);

        AssertTypeEqual(t1, t2);
        AssertTypeNotEqual(t2, t3);
        AssertTypeNotEqual(t3, t4);
        AssertTypeNotEqual(t4, t5);
        AssertTypeEqual(t5, t6);
        AssertTypeNotEqual(t6, t7);
    }

    TEST(TestDateTypes, Attrs) {
        auto t1 = date32();
        auto t2 = date64();

        ASSERT_EQ("date32[day]", t1->to_string());
        ASSERT_EQ("date64[ms]", t2->to_string());

        ASSERT_EQ(32, turbo::checked_cast<const FixedWidthType&>(*t1).bit_width());
        ASSERT_EQ(64, turbo::checked_cast<const FixedWidthType&>(*t2).bit_width());
    }

    TEST(TestTimeType, Equals) {
        Time32Type t0;
        Time32Type t1(TimeUnit::SECOND);
        Time32Type t2(TimeUnit::MILLI);
        Time64Type t3(TimeUnit::MICRO);
        Time64Type t4(TimeUnit::NANO);
        Time64Type t5(TimeUnit::MICRO);

        ASSERT_EQ(32, t0.bit_width());
        ASSERT_EQ(64, t3.bit_width());

        AssertTypeEqual(t0, t2);
        AssertTypeEqual(t1, t1);
        AssertTypeNotEqual(t1, t3);
        AssertTypeNotEqual(t3, t4);
        AssertTypeEqual(t3, t5);
    }

    TEST(TestTimeType, ToString) {
        auto t1 = time32(TimeUnit::MILLI);
        auto t2 = time64(TimeUnit::NANO);
        auto t3 = time32(TimeUnit::SECOND);
        auto t4 = time64(TimeUnit::MICRO);

        ASSERT_EQ("time32[ms]", t1->to_string());
        ASSERT_EQ("time64[ns]", t2->to_string());
        ASSERT_EQ("time32[s]", t3->to_string());
        ASSERT_EQ("time64[us]", t4->to_string());
    }

    TEST(TestMonthIntervalType, Equals) {
        MonthIntervalType t1;
        MonthIntervalType t2;
        DayTimeIntervalType t3;

        AssertTypeEqual(t1, t2);
        AssertTypeNotEqual(t1, t3);
    }

    TEST(TestMonthIntervalType, ToString) {
        auto t1 = month_interval();

        ASSERT_EQ("month_interval", t1->to_string());
    }

    TEST(TestDayTimeIntervalType, Equals) {
        DayTimeIntervalType t1;
        DayTimeIntervalType t2;
        MonthIntervalType t3;

        AssertTypeEqual(t1, t2);
        AssertTypeNotEqual(t1, t3);
    }

    TEST(TestDayTimeIntervalType, ToString) {
        auto t1 = day_time_interval();

        ASSERT_EQ("day_time_interval", t1->to_string());
    }

    TEST(TestMonthDayNanoIntervalType, Equals) {
        MonthDayNanoIntervalType t1;
        MonthDayNanoIntervalType t2;
        MonthIntervalType t3;
        DayTimeIntervalType t4;

        AssertTypeEqual(t1, t2);
        AssertTypeNotEqual(t1, t3);
        AssertTypeNotEqual(t1, t4);
    }

    TEST(TestMonthDayNanoIntervalType, ToString) {
        auto t1 = month_day_nano_interval();

        ASSERT_EQ("month_day_nano_interval", t1->to_string());
    }

    TEST(TestDurationType, Equals) {
        DurationType t1;
        DurationType t2;
        DurationType t3(TimeUnit::NANO);
        DurationType t4(TimeUnit::NANO);

        AssertTypeEqual(t1, t2);
        AssertTypeNotEqual(t1, t3);
        AssertTypeEqual(t3, t4);
    }

    TEST(TestDurationType, ToString) {
        auto t1 = duration(TimeUnit::MILLI);
        auto t2 = duration(TimeUnit::NANO);
        auto t3 = duration(TimeUnit::SECOND);
        auto t4 = duration(TimeUnit::MICRO);

        ASSERT_EQ("duration[ms]", t1->to_string());
        ASSERT_EQ("duration[ns]", t2->to_string());
        ASSERT_EQ("duration[s]", t3->to_string());
        ASSERT_EQ("duration[us]", t4->to_string());
    }

    TEST(TestTimestampType, Equals) {
        TimestampType t1;
        TimestampType t2;
        TimestampType t3(TimeUnit::NANO);
        TimestampType t4(TimeUnit::NANO);

        DurationType dt1;
        DurationType dt2(TimeUnit::NANO);

        AssertTypeEqual(t1, t2);
        AssertTypeNotEqual(t1, t3);
        AssertTypeEqual(t3, t4);

        AssertTypeNotEqual(t1, dt1);
        AssertTypeNotEqual(t3, dt2);
    }

    TEST(TestTimestampType, ToString) {
        auto t1 = timestamp(TimeUnit::MILLI);
        auto t2 = timestamp(TimeUnit::NANO, "US/Eastern");
        auto t3 = timestamp(TimeUnit::SECOND);
        auto t4 = timestamp(TimeUnit::MICRO);

        ASSERT_EQ("timestamp[ms]", t1->to_string());
        ASSERT_EQ("timestamp[ns, tz=US/Eastern]", t2->to_string());
        ASSERT_EQ("timestamp[s]", t3->to_string());
        ASSERT_EQ("timestamp[us]", t4->to_string());
    }

    TEST(TestListType, Equals) {
        auto t1 = list(utf8());
        auto t2 = list(utf8());
        auto t3 = list(binary());
        auto t4 = list(field("item", utf8(), /*nullable=*/false));
        auto tl1 = large_list(binary());
        auto tl2 = large_list(binary());
        auto tl3 = large_list(float64());

        AssertTypeEqual(*t1, *t2);
        AssertTypeNotEqual(*t1, *t3);
        AssertTypeNotEqual(*t1, *t4);
        AssertTypeNotEqual(*t3, *tl1);
        AssertTypeEqual(*tl1, *tl2);
        AssertTypeNotEqual(*tl2, *tl3);

        std::shared_ptr<DataType> vt = std::make_shared<UInt8Type>();
        std::shared_ptr<Field> inner_field = std::make_shared<Field>("non_default_name", vt);

        ListType list_type(vt);
        ListType list_type_named(inner_field);

        AssertTypeEqual(list_type, list_type_named);
        ASSERT_FALSE(list_type.equals(list_type_named, /*check_metadata=*/true));
    }

    TEST(TestListViewType, Equals) {
        auto t1 = list_view(utf8());
        auto t2 = list_view(utf8());
        auto t3 = list_view(binary());
        auto t4 = list_view(field("item", utf8(), /*nullable=*/false));
        auto tl1 = large_list_view(binary());
        auto tl2 = large_list_view(binary());
        auto tl3 = large_list_view(float64());

        AssertTypeEqual(*t1, *t2);
        AssertTypeNotEqual(*t1, *t3);
        AssertTypeNotEqual(*t1, *t4);
        AssertTypeNotEqual(*t3, *tl1);
        AssertTypeEqual(*tl1, *tl2);
        AssertTypeNotEqual(*tl2, *tl3);

        std::shared_ptr<DataType> vt = std::make_shared<UInt8Type>();
        std::shared_ptr<Field> inner_field = std::make_shared<Field>("non_default_name", vt);

        ListViewType list_view_type(vt);
        ListViewType list_view_type_named(inner_field);

        AssertTypeEqual(list_view_type, list_view_type_named);
        ASSERT_FALSE(list_view_type.equals(list_view_type_named, /*check_metadata=*/true));
        ASSERT_NE(list_view_type.to_string(), list_view_type_named.to_string());
    }

    using ListListTypeFactory =
    std::function<std::shared_ptr<DataType>(std::shared_ptr<Field>)>;

    void CheckListListTypeMetadata(ListListTypeFactory list_type_factory) {
        auto md1 = key_value_metadata({"foo", "bar"}, {"foo value", "bar value"});
        auto md2 = key_value_metadata({"foo", "bar"}, {"foo value", "bar value"});
        auto md3 = key_value_metadata({"foo"}, {"foo value"});

        auto f1 = field("item", utf8(), /*nullable =*/true, md1);
        auto f2 = field("item", utf8(), /*nullable =*/true, md2);
        auto f3 = field("item", utf8(), /*nullable =*/true, md3);
        auto f4 = field("item", utf8());
        auto f5 = field("item", utf8(), /*nullable =*/false, md1);

        auto t1 = list_type_factory(f1);
        auto t2 = list_type_factory(f2);
        auto t3 = list_type_factory(f3);
        auto t4 = list_type_factory(f4);
        auto t5 = list_type_factory(f5);

        AssertTypeEqual(*t1, *t2);
        AssertTypeEqual(*t1, *t2, /*check_metadata =*/false);
        ASSERT_EQ(t1->to_string(/*show_metadata=*/true), t2->to_string(/*show_metadata=*/true));

        AssertTypeEqual(*t1, *t3);
        AssertTypeNotEqual(*t1, *t3, /*check_metadata =*/true);
        ASSERT_EQ(t1->to_string(/*show_metadata=*/false), t3->to_string(/*show_metadata=*/false));
        ASSERT_NE(t1->to_string(/*show_metadata=*/true), t3->to_string(/*show_metadata=*/true));

        AssertTypeEqual(*t1, *t4);
        AssertTypeNotEqual(*t1, *t4, /*check_metadata =*/true);
        ASSERT_EQ(t1->to_string(/*show_metadata=*/false), t4->to_string(/*show_metadata=*/false));
        ASSERT_NE(t1->to_string(/*show_metadata=*/true), t4->to_string(/*show_metadata=*/true));

        AssertTypeNotEqual(*t1, *t5);
        AssertTypeNotEqual(*t1, *t5, /*check_metadata =*/true);
        ASSERT_NE(t1->to_string(/*show_metadata=*/false), t5->to_string(/*show_metadata=*/false));
        ASSERT_NE(t1->to_string(/*show_metadata=*/true), t5->to_string(/*show_metadata=*/true));
    }

    TEST(TestListType, Metadata) {
        CheckListListTypeMetadata([](std::shared_ptr<Field> field) { return list(field); });
    }

    TEST(TestLargeListType, Metadata) {
        CheckListListTypeMetadata(
            [](std::shared_ptr<Field> field) { return large_list(field); });
    }

    TEST(TestListViewType, Metadata) {
        CheckListListTypeMetadata(
            [](std::shared_ptr<Field> field) { return list_view(field); });
    }

    TEST(TestLargeListViewType, Metadata) {
        CheckListListTypeMetadata(
            [](std::shared_ptr<Field> field) { return large_list_view(field); });
    }

    TEST(TestNestedType, Equals) {
        auto create_struct = [](std::string inner_name,
                                std::string struct_name) -> std::shared_ptr<Field> {
            auto f_type = field(inner_name, int32());
            std::vector<std::shared_ptr<Field> > fields = {f_type};
            auto s_type = std::make_shared<StructType>(fields);
            return field(struct_name, s_type);
        };

        auto create_union = [](std::string inner_name,
                               std::string union_name) -> std::shared_ptr<Field> {
            auto f_type = field(inner_name, int32());
            std::vector<std::shared_ptr<Field> > fields = {f_type};
            std::vector<int8_t> codes = {42};
            return field(union_name, sparse_union(fields, codes));
        };

        auto s0 = create_struct("f0", "s0");
        auto s0_other = create_struct("f0", "s0");
        auto s0_bad = create_struct("f1", "s0");
        auto s1 = create_struct("f1", "s1");

        AssertFieldEqual(*s0, *s0_other);
        AssertFieldNotEqual(*s0, *s1);
        AssertFieldNotEqual(*s0, *s0_bad);

        auto u0 = create_union("f0", "u0");
        auto u0_other = create_union("f0", "u0");
        auto u0_bad = create_union("f1", "u0");
        auto u1 = create_union("f1", "u1");

        AssertFieldEqual(*u0, *u0_other);
        AssertFieldNotEqual(*u0, *u1);
        AssertFieldNotEqual(*u0, *u0_bad);
    }

    TEST(TestStructType, Basics) {
        auto f0_type = int32();
        auto f0 = field("f0", f0_type);

        auto f1_type = utf8();
        auto f1 = field("f1", f1_type);

        auto f2_type = uint8();
        auto f2 = field("f2", f2_type);

        std::vector<std::shared_ptr<Field> > fields = {f0, f1, f2};

        StructType struct_type(fields);

        ASSERT_TRUE(struct_type.field(0)->equals(f0));
        ASSERT_TRUE(struct_type.field(1)->equals(f1));
        ASSERT_TRUE(struct_type.field(2)->equals(f2));

        ASSERT_EQ(struct_type.to_string(), "struct<f0: int32, f1: string, f2: uint8>");

        auto t1 = STRUCT({{"a", int8()}, {"b", utf8()}});
        auto t2 = STRUCT({field("a", int8()), field("b", utf8())});
        auto t3 = STRUCT({field("c", int8()), field("b", utf8())});
        ASSERT_TRUE(t1->equals(t2));
        ASSERT_TRUE(!t1->equals(t3));

        // TODO(wesm): out of bounds for field(...)
    }

    TEST(TestStructType, get_field_by_name) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f3", list(int16()));

        StructType struct_type({f0, f1, f2, f3});
        std::shared_ptr<Field> result;

        result = struct_type.get_field_by_name("f1");
        ASSERT_EQ(f1, result);

        result = struct_type.get_field_by_name("f3");
        ASSERT_EQ(f3, result);

        result = struct_type.get_field_by_name("not-found");
        ASSERT_EQ(result, nullptr);
    }

    TEST(TestStructType, get_field_index) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", uint8(), false);
        auto f2 = field("f2", utf8());
        auto f3 = field("f3", list(int16()));

        StructType struct_type({f0, f1, f2, f3});

        ASSERT_EQ(0, struct_type.get_field_index(f0->name()));
        ASSERT_EQ(1, struct_type.get_field_index(f1->name()));
        ASSERT_EQ(2, struct_type.get_field_index(f2->name()));
        ASSERT_EQ(3, struct_type.get_field_index(f3->name()));
        ASSERT_EQ(-1, struct_type.get_field_index("not-found"));
    }

    TEST(TestStructType, GetFieldDuplicates) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", int64());
        auto f2 = field("f1", utf8());
        StructType struct_type({f0, f1, f2});

        ASSERT_EQ(0, struct_type.get_field_index("f0"));
        ASSERT_EQ(-1, struct_type.get_field_index("f1"));
        ASSERT_EQ(std::vector<int>{0}, struct_type.get_all_field_indices(f0->name()));
        ASSERT_EQ(std::vector<int>({1, 2}), struct_type.get_all_field_indices(f1->name()));

        std::vector<std::shared_ptr<Field> > results;

        results = struct_type.get_all_fields_by_name(f0->name());
        ASSERT_EQ(results.size(), 1);
        ASSERT_TRUE(results[0]->equals(f0));

        results = struct_type.get_all_fields_by_name(f1->name());
        ASSERT_EQ(results.size(), 2);
        if (results[0]->type()->id() == Type::INT64) {
            ASSERT_TRUE(results[0]->equals(f1));
            ASSERT_TRUE(results[1]->equals(f2));
        } else {
            ASSERT_TRUE(results[0]->equals(f2));
            ASSERT_TRUE(results[1]->equals(f1));
        }

        results = struct_type.get_all_fields_by_name("not-found");
        ASSERT_EQ(results.size(), 0);
    }

    TEST(TestStructType, TestFieldsDifferOnlyInMetadata) {
        auto f0 = field("f", utf8(), true, nullptr);
        auto f1 = field("f", utf8(), true, key_value_metadata({{"foo", "baz"}}));

        StructType s0({f0, f1});
        StructType s1({f1, f0});

        AssertTypeEqual(s0, s1);
        AssertTypeNotEqual(s0, s1, /* check_metadata = */ true);
        ASSERT_NE(s0.to_string(), s1.to_string(/*show_metadata=*/true));

        std::string expected = R"(struct<f: string
-- metadata --
foo: baz, f: string>)";
        ASSERT_EQ(s1.to_string(/*show_metadata=*/true), expected);

        ASSERT_EQ(s0.fingerprint(), s1.fingerprint());
        ASSERT_NE(s0.metadata_fingerprint(), s1.metadata_fingerprint());
    }

    TEST(TestStructType, FieldModifierMethods) {
        auto f0 = field("f0", int32());
        auto f1 = field("f1", utf8());

        std::vector<std::shared_ptr<Field> > fields = {f0, f1};

        StructType struct_type(fields);

        ASSERT_OK_AND_ASSIGN(auto new_struct, struct_type.add_field(1, field("f2", int8())));
        ASSERT_EQ(3, new_struct->num_fields());
        ASSERT_EQ(1, new_struct->get_field_index("f2"));

        ASSERT_OK_AND_ASSIGN(new_struct, new_struct->remove_field(1));
        ASSERT_EQ(2, new_struct->num_fields());
        ASSERT_EQ(-1, new_struct->get_field_index("f2")); // No f2 after removal

        ASSERT_OK_AND_ASSIGN(new_struct, new_struct->set_field(1, field("f2", int8())));
        ASSERT_EQ(2, new_struct->num_fields());
        ASSERT_EQ(1, new_struct->get_field_index("f2"));
        ASSERT_EQ(int8(), new_struct->get_field_by_name("f2")->type());

        EXPECT_RAISES_WITH_MESSAGE_THAT(turbo::StatusCode::kInvalidArgument,
                                        testing::HasSubstr("Invalid column index to add field"),
                                        new_struct->add_field(5, field("f5", int8())));
        EXPECT_RAISES_WITH_MESSAGE_THAT(
            turbo::StatusCode::kInvalidArgument, testing::HasSubstr("Invalid column index to remove field"),
            new_struct->remove_field(-1));
        EXPECT_RAISES_WITH_MESSAGE_THAT(turbo::StatusCode::kInvalidArgument,
                                        testing::HasSubstr("Invalid column index to set field"),
                                        new_struct->set_field(2, field("f5", int8())));
    }

    TEST(TestUnionType, Basics) {
        auto f0_type = int32();
        auto f0 = field("f0", f0_type);
        auto f1_type = utf8();
        auto f1 = field("f1", f1_type);
        auto f2_type = uint8();
        auto f2 = field("f2", f2_type);

        std::vector<std::shared_ptr<Field> > fields = {f0, f1, f2};
        std::vector<int8_t> type_codes1 = {0, 1, 2};
        std::vector<int8_t> type_codes2 = {10, 11, 12};
        std::vector<int> child_ids1(128, -1);
        std::vector<int> child_ids2(128, -1);
        child_ids1[0] = 0;
        child_ids1[1] = 1;
        child_ids1[2] = 2;
        child_ids2[10] = 0;
        child_ids2[11] = 1;
        child_ids2[12] = 2;

        auto ty1 = turbo::checked_pointer_cast<UnionType>(dense_union(fields));
        auto ty2 = turbo::checked_pointer_cast<UnionType>(dense_union(fields, type_codes1));
        auto ty3 = turbo::checked_pointer_cast<UnionType>(dense_union(fields, type_codes2));
        auto ty4 = turbo::checked_pointer_cast<UnionType>(sparse_union(fields));
        auto ty5 = turbo::checked_pointer_cast<UnionType>(sparse_union(fields, type_codes1));
        auto ty6 = turbo::checked_pointer_cast<UnionType>(sparse_union(fields, type_codes2));

        ASSERT_EQ(ty1->type_codes(), type_codes1);
        ASSERT_EQ(ty2->type_codes(), type_codes1);
        ASSERT_EQ(ty3->type_codes(), type_codes2);
        ASSERT_EQ(ty4->type_codes(), type_codes1);
        ASSERT_EQ(ty5->type_codes(), type_codes1);
        ASSERT_EQ(ty6->type_codes(), type_codes2);

        ASSERT_EQ(ty1->child_ids(), child_ids1);
        ASSERT_EQ(ty2->child_ids(), child_ids1);
        ASSERT_EQ(ty3->child_ids(), child_ids2);
        ASSERT_EQ(ty4->child_ids(), child_ids1);
        ASSERT_EQ(ty5->child_ids(), child_ids1);
        ASSERT_EQ(ty6->child_ids(), child_ids2);
    }

    TEST(TestDictionaryType, Basics) {
        auto value_type = int32();

        std::shared_ptr<DictionaryType> type1 =
                std::dynamic_pointer_cast<DictionaryType>(dictionary(int16(), value_type));

        auto type2 = std::dynamic_pointer_cast<DictionaryType>(
            ::nebula::dictionary(int16(), type1, true));

        ASSERT_TRUE(int16()->equals(type1->index_type()));
        ASSERT_TRUE(type1->get_value_type()->equals(value_type));

        ASSERT_TRUE(int16()->equals(type2->index_type()));
        ASSERT_TRUE(type2->get_value_type()->equals(type1));

        ASSERT_EQ("dictionary<values=int32, indices=int16, ordered=0>", type1->to_string());
        ASSERT_EQ(
            "dictionary<values="
            "dictionary<values=int32, indices=int16, ordered=0>, "
            "indices=int16, ordered=1>",
            type2->to_string());
    }

    TEST(TestDictionaryType, Equals) {
        auto t1 = dictionary(int8(), int32());
        auto t2 = dictionary(int8(), int32());
        auto t3 = dictionary(int16(), int32());
        auto t4 = dictionary(int8(), int16());

        AssertTypeEqual(*t1, *t2);
        AssertTypeNotEqual(*t1, *t3);
        AssertTypeNotEqual(*t1, *t4);

        auto t5 = dictionary(int8(), int32(), /*ordered=*/false);
        auto t6 = dictionary(int8(), int32(), /*ordered=*/true);
        AssertTypeNotEqual(*t5, *t6);
    }

    TEST(TypesTest, TestDecimal128Small) {
        Decimal128Type t1(8, 4);

        EXPECT_EQ(t1.id(), Type::DECIMAL128);
        EXPECT_EQ(t1.precision(), 8);
        EXPECT_EQ(t1.scale(), 4);

        EXPECT_EQ(t1.to_string(), std::string("decimal128(8, 4)"));

        // Test properties
        EXPECT_EQ(t1.byte_width(), 16);
        EXPECT_EQ(t1.bit_width(), 128);
    }

    TEST(TypesTest, TestDecimal128Medium) {
        Decimal128Type t1(12, 5);

        EXPECT_EQ(t1.id(), Type::DECIMAL128);
        EXPECT_EQ(t1.precision(), 12);
        EXPECT_EQ(t1.scale(), 5);

        EXPECT_EQ(t1.to_string(), std::string("decimal128(12, 5)"));

        // Test properties
        EXPECT_EQ(t1.byte_width(), 16);
        EXPECT_EQ(t1.bit_width(), 128);
    }

    TEST(TypesTest, TestDecimal128Large) {
        Decimal128Type t1(27, 7);

        EXPECT_EQ(t1.id(), Type::DECIMAL128);
        EXPECT_EQ(t1.precision(), 27);
        EXPECT_EQ(t1.scale(), 7);

        EXPECT_EQ(t1.to_string(), std::string("decimal128(27, 7)"));

        // Test properties
        EXPECT_EQ(t1.byte_width(), 16);
        EXPECT_EQ(t1.bit_width(), 128);
    }

    TEST(TypesTest, TestDecimal256Small) {
        Decimal256Type t1(8, 4);

        EXPECT_EQ(t1.id(), Type::DECIMAL256);
        EXPECT_EQ(t1.precision(), 8);
        EXPECT_EQ(t1.scale(), 4);

        EXPECT_EQ(t1.to_string(), std::string("decimal256(8, 4)"));

        // Test properties
        EXPECT_EQ(t1.byte_width(), 32);
        EXPECT_EQ(t1.bit_width(), 256);
    }

    TEST(TypesTest, TestDecimal256Medium) {
        Decimal256Type t1(12, 5);

        EXPECT_EQ(t1.id(), Type::DECIMAL256);
        EXPECT_EQ(t1.precision(), 12);
        EXPECT_EQ(t1.scale(), 5);

        EXPECT_EQ(t1.to_string(), std::string("decimal256(12, 5)"));

        // Test properties
        EXPECT_EQ(t1.byte_width(), 32);
        EXPECT_EQ(t1.bit_width(), 256);
    }

    TEST(TypesTest, TestDecimal256Large) {
        Decimal256Type t1(76, 38);

        EXPECT_EQ(t1.id(), Type::DECIMAL256);
        EXPECT_EQ(t1.precision(), 76);
        EXPECT_EQ(t1.scale(), 38);

        EXPECT_EQ(t1.to_string(), std::string("decimal256(76, 38)"));

        // Test properties
        EXPECT_EQ(t1.byte_width(), 32);
        EXPECT_EQ(t1.bit_width(), 256);
    }

    TEST(TypesTest, TestDecimalEquals) {
        Decimal128Type t1(8, 4);
        Decimal128Type t2(8, 4);
        Decimal128Type t3(8, 5);
        Decimal128Type t4(27, 5);

        Decimal256Type t5(8, 4);
        Decimal256Type t6(8, 4);
        Decimal256Type t7(8, 5);
        Decimal256Type t8(27, 5);

        FixedSizeBinaryType t9(16);
        FixedSizeBinaryType t10(32);

        AssertTypeEqual(t1, t2);
        AssertTypeNotEqual(t1, t3);
        AssertTypeNotEqual(t1, t4);
        AssertTypeNotEqual(t1, t9);

        AssertTypeEqual(t5, t6);
        AssertTypeNotEqual(t5, t1);
        AssertTypeNotEqual(t5, t7);
        AssertTypeNotEqual(t5, t8);
        AssertTypeNotEqual(t5, t10);
    }

    TEST(TypesTest, TestRunEndEncodedType) {
        auto int8_ree_expected = std::make_shared<RunEndEncodedType>(int32(), list(int8()));
        auto int8_ree_type = run_end_encoded(int32(), list(int8()));
        auto int32_ree_type = run_end_encoded(int32(), list(int32()));

        ASSERT_EQ(*int8_ree_expected, *int8_ree_type);
        ASSERT_NE(*int8_ree_expected, *int32_ree_type);

        ASSERT_EQ(int8_ree_type->id(), Type::RUN_END_ENCODED);
        ASSERT_EQ(int32_ree_type->id(), Type::RUN_END_ENCODED);

        auto int8_ree_type_cast = std::dynamic_pointer_cast<RunEndEncodedType>(int8_ree_type);
        auto int32_ree_type_cast = std::dynamic_pointer_cast<RunEndEncodedType>(int32_ree_type);
        ASSERT_EQ(*int8_ree_type_cast->get_value_type(), *list(int8()));
        ASSERT_EQ(*int32_ree_type_cast->get_value_type(), *list(int32()));

        ASSERT_TRUE(int8_ree_type_cast->field(0)->equals(Field("run_ends", int32(), false)));
        ASSERT_TRUE(int8_ree_type_cast->field(1)->equals(Field("values", list(int8()), true)));

        auto int16_int32_ree_type = run_end_encoded(int16(), list(int32()));
        auto int64_int32_ree_type = run_end_encoded(int64(), list(int32()));
        ASSERT_NE(*int32_ree_type, *int16_int32_ree_type);
        ASSERT_NE(*int32_ree_type, *int64_int32_ree_type);
        ASSERT_NE(*int16_int32_ree_type, *int64_int32_ree_type);

        ASSERT_EQ(int16_int32_ree_type->to_string(),
                  "run_end_encoded<run_ends: int16, values: list<item: int32>>");
        ASSERT_EQ(int8_ree_type->to_string(),
                  "run_end_encoded<run_ends: int32, values: list<item: int8>>");
        ASSERT_EQ(int64_int32_ree_type->to_string(),
                  "run_end_encoded<run_ends: int64, values: list<item: int32>>");
    }

    TEST(TypesTest, TestListViewType) {
        auto int32_expected = std::make_shared<ListViewType>(int32());
        auto int32_list_view_type = list_view(int32());

        ASSERT_EQ(*int32_expected, *int32_list_view_type);

        auto int32_list_view_type_cast =
                std::dynamic_pointer_cast<ListViewType>(int32_list_view_type);
        ASSERT_EQ(*int32_list_view_type_cast->get_value_type(), *int32());

        ASSERT_TRUE(int32_list_view_type->field(0)->equals(Field("item", int32(), true)));

        auto int64_list_view_type = list_view(int64());
        ASSERT_NE(*int32_list_view_type, *int64_list_view_type);

        ASSERT_EQ(int32_list_view_type->to_string(), "list_view<item: int32>");
        ASSERT_EQ(int64_list_view_type->to_string(), "list_view<item: int64>");
    }

    TEST(TypesTest, TestLargeListViewType) {
        auto int32_expected = std::make_shared<LargeListViewType>(int32());
        auto int32_list_view_type = large_list_view(int32());

        ASSERT_EQ(*int32_expected, *int32_list_view_type);

        auto int32_list_view_type_cast =
                std::dynamic_pointer_cast<LargeListViewType>(int32_list_view_type);
        ASSERT_EQ(*int32_list_view_type_cast->get_value_type(), *int32());

        ASSERT_TRUE(int32_list_view_type->field(0)->equals(Field("item", int32(), true)));

        auto int64_list_view_type = large_list_view(int64());
        ASSERT_NE(*int32_list_view_type, *int64_list_view_type);

        ASSERT_EQ(int32_list_view_type->to_string(), "large_list_view<item: int32>");
        ASSERT_EQ(int64_list_view_type->to_string(), "large_list_view<item: int64>");
    }

#define TEST_PREDICATE(all_types, type_predicate)                 \
  for (auto type : all_types) {                                   \
    ASSERT_EQ(type_predicate(type->id()), type_predicate(*type)); \
  }

    TEST(TypesTest, TestMembership) {
        std::vector<std::shared_ptr<DataType> > all_types;
        for (auto type: numeric_types()) {
            all_types.push_back(type);
        }
        for (auto type: temporal_types()) {
            all_types.push_back(type);
        }
        for (auto type: interval_types()) {
            all_types.push_back(type);
        }
        for (auto type: primitive_types()) {
            all_types.push_back(type);
        }
        TEST_PREDICATE(all_types, is_integer);
        TEST_PREDICATE(all_types, is_signed_integer);
        TEST_PREDICATE(all_types, is_unsigned_integer);
        TEST_PREDICATE(all_types, is_floating);
        TEST_PREDICATE(all_types, is_numeric);
        TEST_PREDICATE(all_types, is_decimal);
        TEST_PREDICATE(all_types, is_primitive);
        TEST_PREDICATE(all_types, is_base_binary_like);
        TEST_PREDICATE(all_types, is_binary_like);
        TEST_PREDICATE(all_types, is_large_binary_like);
        TEST_PREDICATE(all_types, is_binary);
        TEST_PREDICATE(all_types, is_string);
        TEST_PREDICATE(all_types, is_temporal);
        TEST_PREDICATE(all_types, is_interval);
        TEST_PREDICATE(all_types, is_dictionary);
        TEST_PREDICATE(all_types, is_fixed_size_binary);
        TEST_PREDICATE(all_types, is_fixed_width);
        TEST_PREDICATE(all_types, is_var_length_list);
        TEST_PREDICATE(all_types, is_list_like);
        TEST_PREDICATE(all_types, is_var_length_list_like);
        TEST_PREDICATE(all_types, is_nested);
        TEST_PREDICATE(all_types, is_union);
    }

#undef TEST_PREDICATE
} // namespace nebula
